rchd 2.1.3

//! Worker pool management.

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

use crate::DaemonContext;
use crate::disk_pressure::{
    DiskPressurePolicyConfig, DiskSlotPolicy, PressureAssessment, evaluate_pressure_policy,
};
use crate::health::probe_worker_capabilities;
use rch_common::{
    CircuitBreakerConfig, CircuitState, CircuitStats, WorkerCapabilities, WorkerConfig, WorkerId,
    WorkerStatus,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicI64, AtomicU32, AtomicU64, AtomicUsize, Ordering};
use std::time::Duration;
use tokio::sync::RwLock;
use tracing::debug;

// Sized for a *loaded* worker, not an idle one.
//
// `rch-wkr capabilities` shells out to rustc/node/npm/go/zig/cargo-zigbuild and
// stats the disk, behind a fresh SSH handshake. On an idle box that is ~1s; on a
// busy build host it is far more. Measured across the fleet with builds running
// (10-core workers at load 45-60): 2.9s, 6.3s, 7.2s, 8.6s, **15.8s**. Against the
// former 4s/8s pair every one of those hosts failed its probe.
//
// That failure is self-defeating: a probe that only fails under load means
// `disk_free_gb` goes stale exactly on the workers that are actively filling
// their disks, and `is_low_disk()` returns `None` for missing data — so the
// low-disk admission gate FAILS OPEN and the dispatcher keeps feeding a worker
// that is about to hit ENOSPC.
//
// This is a low-frequency operator query (`rch workers capabilities --refresh`),
// not one of the ~30s poll loops, and refreshes run concurrently across workers,
// so a generous ceiling costs nothing in the common case and only ever shortens
// the tail.
const CAPABILITIES_REFRESH_PROBE_TIMEOUT: Duration = Duration::from_secs(25);
const CAPABILITIES_REFRESH_WORKER_BUDGET: Duration = Duration::from_secs(35);

fn duration_millis_i64(duration: Duration) -> i64 {
    i64::try_from(duration.as_millis()).unwrap_or(i64::MAX)
}

fn duration_secs_i64(duration: Duration) -> i64 {
    i64::try_from(duration.as_secs()).unwrap_or(i64::MAX)
}

// =============================================================================
// Worker lifecycle type model (bd-session-history-remediation-ocv9i.1.1)
// =============================================================================
//
// A worker's lifecycle is modelled as **two independent axes** so that
// permanent operator intent can never be confused with transient scheduler
// eligibility:
//
//   1. [`AdminIntent`]  — the *desired inventory* axis. Changes ONLY through
//      explicit operator actions (`rch workers disable/enable/drain`). This is
//      what `workers.toml` encodes; it is the source of truth for what the
//      operator wants in the fleet.
//   2. [`EligibilityState`] — the *live scheduler eligibility* axis. Advances
//      automatically from health probes and concrete failure classes. It NEVER
//      writes back to `AdminIntent` / `workers.toml`.
//
// The cardinal safety invariant (session-history remediation): a transient
// failure quarantines a worker on the eligibility axis (`TemporaryBypass`) but
// must never mutate desired inventory, and a recovering worker must never
// auto-rejoin if the operator has disabled it. Keeping the two axes as distinct
// types makes "auto-rejoin an operator-disabled worker" unrepresentable rather
// than merely discouraged.
//
// ## Serialization compatibility policy
//
// These types derive `serde` with `snake_case` so they are stable across the
// daemon/status JSON boundary. They are *additive*: the legacy single-axis
// [`WorkerStatus`] enum is untouched, and [`WorkerLifecycle::legacy_status`]
// collapses the two axes back to a `WorkerStatus` for older status consumers
// (both quarantine states — `TemporaryBypass` and `RecoveredPendingCanary` —
// collapse to `Unreachable` so they stay out of legacy scheduling).
// New variants must be appended, never reordered or renamed, so historical
// JSONL replays continue to deserialize.

/// Operator-controlled *desired* lifecycle intent for a worker.
///
/// This is the desired-inventory axis. It changes only through explicit
/// operator actions and is never advanced by health probes or transient
/// failures. A worker is only ever schedulable when its intent is
/// [`AdminIntent::Active`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AdminIntent {
    /// Operator wants this worker in service (default).
    #[default]
    Active,
    /// Operator asked the worker to stop taking *new* jobs but finish current
    /// ones (`rch workers drain`).
    Draining,
    /// Drain has completed (no active jobs); the operator's intent is still
    /// "out of service" until they re-enable.
    Drained,
    /// Operator explicitly disabled the worker. Never eligible — and never
    /// auto-rejoined by the recovery path — until re-enabled.
    Disabled,
}

/// Transient *scheduler eligibility* derived from health probes and failure
/// classes.
///
/// This is the live axis. It advances automatically and never writes back to
/// [`AdminIntent`]. The quarantine lifecycle is:
///
/// ```text
/// Healthy/Degraded/Unreachable --enter_bypass(class)--> TemporaryBypass
/// TemporaryBypass              --recover_to_canary----> RecoveredPendingCanary
/// RecoveredPendingCanary       --promote_from_canary--> Healthy
/// RecoveredPendingCanary       --enter_bypass(class)--> TemporaryBypass   (relapse)
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EligibilityState {
    /// Healthy and accepting work.
    #[default]
    Healthy,
    /// Responding but slow/partially degraded — still eligible for work.
    Degraded,
    /// Failed to respond to heartbeat — temporarily ineligible, but not yet
    /// quarantined into a failure-class bypass.
    Unreachable,
    /// Quarantined out of scheduling because a probe hit a concrete failure
    /// class. Distinct from [`AdminIntent::Disabled`]: this is transient and
    /// recovers automatically via probe + canary.
    TemporaryBypass,
    /// A previously bypassed worker passed its recovery probe and is awaiting a
    /// single canary build before full rejoin.
    RecoveredPendingCanary,
}

/// Concrete failure class that quarantined a worker into
/// [`EligibilityState::TemporaryBypass`]. Defined in `rch_common` so the shared
/// bypass-record schema, incident ledger, and status surfaces speak one
/// vocabulary; re-exported here as it is part of the worker lifecycle model.
pub use rch_common::BypassFailureClass;

/// An attempted lifecycle transition that the state machine rejects.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IllegalLifecycleTransition {
    /// Eligibility state the worker was in.
    pub from: EligibilityState,
    /// Eligibility state that was requested.
    pub to: EligibilityState,
    /// Why the transition is not permitted.
    pub reason: &'static str,
}

impl std::fmt::Display for IllegalLifecycleTransition {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "illegal worker lifecycle transition {:?} -> {:?}: {}",
            self.from, self.to, self.reason
        )
    }
}

impl std::error::Error for IllegalLifecycleTransition {}

/// The full lifecycle state of a worker: the two independent axes plus the
/// failure class that explains a current bypass (if any).
///
/// Scheduling decisions combine both axes via [`WorkerLifecycle::is_schedulable`]
/// — admin intent dominates, but neither axis ever overwrites the other.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
pub struct WorkerLifecycle {
    /// Operator-controlled desired intent.
    pub admin: AdminIntent,
    /// Health-/probe-driven live eligibility.
    pub eligibility: EligibilityState,
    /// Set iff `eligibility == TemporaryBypass`; explains the quarantine.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub bypass_cause: Option<BypassFailureClass>,
}

impl WorkerLifecycle {
    /// A fresh, fully in-service worker: `Active` + `Healthy`.
    pub fn new() -> Self {
        Self {
            admin: AdminIntent::Active,
            eligibility: EligibilityState::Healthy,
            bypass_cause: None,
        }
    }

    /// Bridge from the legacy single-axis [`WorkerStatus`] so existing call
    /// sites can be migrated incrementally. Admin-intent statuses populate the
    /// admin axis with a `Healthy` eligibility; health statuses populate the
    /// eligibility axis with an `Active` intent.
    pub fn from_worker_status(status: WorkerStatus) -> Self {
        match status {
            WorkerStatus::Healthy => Self::new(),
            WorkerStatus::Degraded => Self {
                admin: AdminIntent::Active,
                eligibility: EligibilityState::Degraded,
                bypass_cause: None,
            },
            WorkerStatus::Unreachable => Self {
                admin: AdminIntent::Active,
                eligibility: EligibilityState::Unreachable,
                bypass_cause: None,
            },
            WorkerStatus::Draining => Self {
                admin: AdminIntent::Draining,
                eligibility: EligibilityState::Healthy,
                bypass_cause: None,
            },
            WorkerStatus::Drained => Self {
                admin: AdminIntent::Drained,
                eligibility: EligibilityState::Healthy,
                bypass_cause: None,
            },
            WorkerStatus::Disabled => Self {
                admin: AdminIntent::Disabled,
                eligibility: EligibilityState::Healthy,
                bypass_cause: None,
            },
        }
    }

    /// Collapse the two axes back to a legacy [`WorkerStatus`] for older
    /// status/JSON consumers. Admin intent takes precedence.
    ///
    /// Both transient quarantine states collapse to `Unreachable` so they stay
    /// EXCLUDED from the legacy scheduler (which treats `Degraded` as
    /// schedulable). In particular `RecoveredPendingCanary` must NOT become
    /// `Degraded`: the legacy scheduler would then route a normal build to a
    /// worker that is only cleared for a single canary build, violating the
    /// one-canary-first invariant that [`Self::is_schedulable`] enforces
    /// (bd-review-canary-legacy-status).
    pub fn legacy_status(&self) -> WorkerStatus {
        match self.admin {
            AdminIntent::Disabled => WorkerStatus::Disabled,
            AdminIntent::Draining => WorkerStatus::Draining,
            AdminIntent::Drained => WorkerStatus::Drained,
            AdminIntent::Active => match self.eligibility {
                EligibilityState::Healthy => WorkerStatus::Healthy,
                EligibilityState::Degraded => WorkerStatus::Degraded,
                // Unreachable, failure-class bypass, and canary-pending all stay
                // out of normal scheduling.
                EligibilityState::Unreachable
                | EligibilityState::TemporaryBypass
                | EligibilityState::RecoveredPendingCanary => WorkerStatus::Unreachable,
            },
        }
    }

    /// Whether the scheduler may route a *normal* build to this worker.
    ///
    /// True only when the operator intends it to be in service (`Active`) and
    /// its live eligibility is `Healthy` or `Degraded`. Unreachable, bypassed,
    /// and canary-pending workers are excluded from normal scheduling.
    pub fn is_schedulable(&self) -> bool {
        self.admin == AdminIntent::Active
            && matches!(
                self.eligibility,
                EligibilityState::Healthy | EligibilityState::Degraded
            )
    }

    /// Whether this worker may receive exactly one canary build (it recovered
    /// from a bypass and the operator still wants it in service). The canary
    /// build itself is driven by a later bead; this predicate gates it.
    pub fn is_canary_pending(&self) -> bool {
        self.admin == AdminIntent::Active
            && self.eligibility == EligibilityState::RecoveredPendingCanary
    }

    /// Apply an operator intent change. This is the *only* way the desired
    /// inventory axis moves; it never touches the eligibility axis.
    pub fn set_admin(&mut self, intent: AdminIntent) {
        self.admin = intent;
    }

    /// Record a plain health observation (`Healthy`/`Degraded`/`Unreachable`).
    ///
    /// Health observations move freely among the three plain health states but
    /// must NOT silently clear a quarantine: if the worker is currently
    /// `TemporaryBypass` or `RecoveredPendingCanary`, the observation is
    /// ignored and `false` is returned. Recovery happens only through the
    /// explicit probe/canary transitions.
    ///
    /// Returns `true` when the observation was applied. Passing a non-health
    /// state (`TemporaryBypass` / `RecoveredPendingCanary`) returns `false`
    /// without mutating; use [`Self::enter_bypass`] / [`Self::recover_to_canary`]
    /// for those.
    pub fn observe_health(&mut self, health: EligibilityState) -> bool {
        let is_plain_health = matches!(
            health,
            EligibilityState::Healthy | EligibilityState::Degraded | EligibilityState::Unreachable
        );
        let currently_quarantined = matches!(
            self.eligibility,
            EligibilityState::TemporaryBypass | EligibilityState::RecoveredPendingCanary
        );
        if !is_plain_health || currently_quarantined {
            return false;
        }
        self.eligibility = health;
        true
    }

    /// Quarantine the worker on the eligibility axis because a probe hit a
    /// concrete failure class. Legal from *any* eligibility state — a failure
    /// is a fact, including a failed canary (`RecoveredPendingCanary` relapses
    /// here) or an already-bypassed worker whose cause is being updated. Never
    /// mutates the admin axis — this is the core "transient failure ≠ desired
    /// inventory" invariant.
    pub fn enter_bypass(&mut self, cause: BypassFailureClass) {
        self.eligibility = EligibilityState::TemporaryBypass;
        self.bypass_cause = Some(cause);
    }

    /// Advance a bypassed worker to `RecoveredPendingCanary` after a successful
    /// recovery probe. Legal only from `TemporaryBypass`.
    pub fn recover_to_canary(&mut self) -> Result<(), IllegalLifecycleTransition> {
        if self.eligibility != EligibilityState::TemporaryBypass {
            return Err(IllegalLifecycleTransition {
                from: self.eligibility,
                to: EligibilityState::RecoveredPendingCanary,
                reason: "recovery to canary is only legal from temporary_bypass",
            });
        }
        self.eligibility = EligibilityState::RecoveredPendingCanary;
        self.bypass_cause = None;
        Ok(())
    }

    /// Fully rejoin a worker after a successful canary build. Legal only from
    /// `RecoveredPendingCanary`. Does not, and must not, change admin intent —
    /// an operator-disabled worker stays disabled even after a clean canary.
    pub fn promote_from_canary(&mut self) -> Result<(), IllegalLifecycleTransition> {
        if self.eligibility != EligibilityState::RecoveredPendingCanary {
            return Err(IllegalLifecycleTransition {
                from: self.eligibility,
                to: EligibilityState::Healthy,
                reason: "promotion is only legal from recovered_pending_canary",
            });
        }
        self.eligibility = EligibilityState::Healthy;
        self.bypass_cause = None;
        Ok(())
    }
}

/// Map a legacy health [`WorkerStatus`] to the eligibility axis for
/// [`WorkerLifecycle::observe_health`]. Administrative statuses
/// (`Draining`/`Drained`/`Disabled`) are not health observations and return
/// `None`, so the caller leaves the lifecycle untouched.
fn health_status_to_eligibility(status: WorkerStatus) -> Option<EligibilityState> {
    match status {
        WorkerStatus::Healthy => Some(EligibilityState::Healthy),
        WorkerStatus::Degraded => Some(EligibilityState::Degraded),
        WorkerStatus::Unreachable => Some(EligibilityState::Unreachable),
        WorkerStatus::Draining | WorkerStatus::Drained | WorkerStatus::Disabled => None,
    }
}

/// Whether a worker may accept a *new* slot reservation given its lifecycle.
///
/// Mirrors the pre-lifecycle `reserve_slots` admin guard (refuse
/// `Draining`/`Drained`/`Disabled`) and additionally refuses the transient
/// quarantine states so a bypassed or canary-pending worker can never have a
/// normal build land on it between selection and reservation. An `Unreachable`
/// but `Active` worker is intentionally still allowed here (unchanged from the
/// legacy guard); the selector already filters it out up front.
fn lifecycle_accepts_new_builds(lifecycle: WorkerLifecycle) -> bool {
    !matches!(
        lifecycle.admin,
        AdminIntent::Draining | AdminIntent::Drained | AdminIntent::Disabled
    ) && !matches!(
        lifecycle.eligibility,
        EligibilityState::TemporaryBypass | EligibilityState::RecoveredPendingCanary
    )
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
enum DrainCompletionAction {
    #[default]
    StayDrained,
    RemoveFromPool,
    Disable {
        reason: Option<String>,
    },
}

/// State of a single worker.
#[derive(Debug)]
pub struct WorkerState {
    /// Worker configuration.
    pub config: RwLock<WorkerConfig>,
    /// Authoritative worker lifecycle — the two-axis (admin intent + live
    /// eligibility) model from [`WorkerLifecycle`].
    ///
    /// **Single source of truth.** The legacy single-axis [`WorkerStatus`] is
    /// *derived* from this via [`WorkerLifecycle::legacy_status`] (see
    /// [`Self::status`]); there is no separate status field that could disagree.
    /// Both transient quarantine states (`TemporaryBypass` /
    /// `RecoveredPendingCanary`) collapse to `Unreachable` for every legacy
    /// consumer, so a bypassed or canary-pending worker stays out of the
    /// scheduler automatically with no extra bookkeeping.
    lifecycle: RwLock<WorkerLifecycle>,
    /// Number of slots currently in use.
    used_slots: Arc<AtomicU32>,
    /// Speed score from benchmarking (0-100).
    ///
    /// **Why atomic:** Read on every worker selection decision (hot path). RwLock's
    /// ~25ns overhead per acquire multiplied by N workers is measurable; AtomicU64
    /// load compiles to a single MOV instruction (~1ns).
    ///
    /// **Bit-casting:** `f64` is stored as `u64` via `f64::to_bits`/`f64::from_bits`.
    /// This is a zero-cost transmute that preserves all IEEE 754 values including
    /// NaN, infinity, and negative zero.
    ///
    /// **Ordering:** `Relaxed` — speed_score is a statistical performance hint, not
    /// a correctness-critical flag. Stale reads produce slightly suboptimal worker
    /// selection, never incorrect behavior. No happens-before dependency with other
    /// fields (status transitions use RwLock which provides its own ordering).
    pub speed_score: AtomicU64,
    /// Last observed worker latency in milliseconds (from health checks).
    ///
    /// **Why atomic:** Same hot-path rationale as `speed_score`.
    ///
    /// **Sentinel:** `0` encodes `None` (latency of exactly 0ms is not meaningful
    /// in practice). Use `last_latency_ms()` accessor which returns `Option<u64>`.
    ///
    /// **Ordering:** `Relaxed` — advisory diagnostic value with no ordering
    /// dependency. See `speed_score` rationale.
    last_latency_ms: AtomicU64,
    /// Projects cached on this worker.
    pub cached_projects: RwLock<Vec<String>>,
    /// Circuit breaker statistics.
    circuit: RwLock<CircuitStats>,
    /// Last error message.
    last_error_msg: RwLock<Option<String>>,
    /// Runtime capabilities (Bun, Node, Rust versions).
    capabilities: RwLock<WorkerCapabilities>,
    /// Serialize daemon-side capability requests from health, operators and selection.
    capability_probe: tokio::sync::Mutex<()>,
    /// Cached per-toolchain preflight verdicts.
    toolchain_preflight: RwLock<HashMap<String, ToolchainPreflightStatus>>,
    /// Latest daemon-side pressure policy assessment for this worker.
    pressure_assessment: RwLock<PressureAssessment>,
    /// Disk budget shared by all workers created by the same daemon pool.
    disk_slot_policy: DiskSlotPolicy,
    /// Reason for disabling this worker (if disabled).
    disabled_reason: RwLock<Option<String>>,
    /// Administrative intent to apply when a draining worker reaches zero slots.
    drain_completion: RwLock<DrainCompletionAction>,
    /// Unix timestamp (seconds since epoch) when worker was disabled.
    ///
    /// **Why atomic:** Queried during worker selection to skip disabled workers.
    ///
    /// **Sentinel:** `0` encodes `None` (no disable event). Use `disabled_at()`
    /// accessor which returns `Option<i64>`.
    ///
    /// **Ordering:** `Relaxed` — diagnostic timestamp set alongside `status`
    /// (RwLock) and `disabled_reason` (RwLock). The RwLock transitions provide
    /// happens-before ordering for the authoritative disabled state; this timestamp
    /// is supplementary. Briefly stale reads are acceptable for display purposes.
    disabled_at: AtomicI64,
}

impl WorkerState {
    /// Create a new worker state from configuration.
    pub fn new(config: WorkerConfig) -> Self {
        Self::with_disk_slot_policy(
            config,
            DiskSlotPolicy::from(&rch_common::SelectionConfig::default()),
        )
    }

    fn with_disk_slot_policy(config: WorkerConfig, disk_slot_policy: DiskSlotPolicy) -> Self {
        Self {
            config: RwLock::new(config),
            lifecycle: RwLock::new(WorkerLifecycle::new()),
            used_slots: Arc::new(AtomicU32::new(0)),
            speed_score: AtomicU64::new(50.0_f64.to_bits()), // Default mid-range score
            last_latency_ms: AtomicU64::new(0),
            cached_projects: RwLock::new(Vec::new()),
            circuit: RwLock::new(CircuitStats::new()),
            last_error_msg: RwLock::new(None),
            capabilities: RwLock::new(WorkerCapabilities::new()),
            capability_probe: tokio::sync::Mutex::new(()),
            toolchain_preflight: RwLock::new(HashMap::new()),
            pressure_assessment: RwLock::new(PressureAssessment::default()),
            disk_slot_policy,
            disabled_reason: RwLock::new(None),
            drain_completion: RwLock::new(DrainCompletionAction::default()),
            disabled_at: AtomicI64::new(0),
        }
    }

    /// Update worker configuration.
    pub async fn update_config(&self, new_config: WorkerConfig) {
        {
            let mut config = self.config.write().await;
            *config = new_config;
        }

        let cancelled_pending_removal = {
            let mut completion = self.drain_completion.write().await;
            let should_cancel = *completion == DrainCompletionAction::RemoveFromPool;
            if should_cancel {
                *completion = DrainCompletionAction::StayDrained;
            }
            should_cancel
        };

        if cancelled_pending_removal {
            let mut lifecycle = self.lifecycle.write().await;
            if matches!(
                lifecycle.admin,
                AdminIntent::Draining | AdminIntent::Drained
            ) {
                lifecycle.set_admin(AdminIntent::Active);
            }
        }
    }

    /// Get the current worker status as the legacy single-axis [`WorkerStatus`].
    ///
    /// This is a *derived* view of the authoritative [`WorkerLifecycle`]: both
    /// quarantine states (`TemporaryBypass` / `RecoveredPendingCanary`) collapse
    /// to [`WorkerStatus::Unreachable`], so a bypassed worker is excluded from
    /// the legacy scheduler with no separate bookkeeping.
    pub async fn status(&self) -> WorkerStatus {
        self.lifecycle.read().await.legacy_status()
    }

    /// Get a copy of the authoritative two-axis [`WorkerLifecycle`].
    pub async fn lifecycle(&self) -> WorkerLifecycle {
        *self.lifecycle.read().await
    }

    /// Get the live scheduler-eligibility axis.
    pub async fn eligibility(&self) -> EligibilityState {
        self.lifecycle.read().await.eligibility
    }

    /// Set worker status from the legacy single-axis [`WorkerStatus`].
    ///
    /// Bridges through [`WorkerLifecycle::from_worker_status`]. This is a legacy
    /// setter for status-API and test call sites; the lifecycle-aware paths
    /// ([`Self::enter_bypass`], [`Self::recover_to_canary`],
    /// [`Self::promote_from_canary`], [`Self::apply_health_status`]) move the
    /// two axes independently and must be used for quarantine/recovery.
    pub async fn set_status(&self, status: WorkerStatus) {
        *self.lifecycle.write().await = WorkerLifecycle::from_worker_status(status);
    }

    /// Apply a health-derived status without overriding administrative lifecycle
    /// states *or* an active quarantine.
    ///
    /// Health probes may mark a worker healthy/degraded/unreachable, but they
    /// must not revive workers that are intentionally `Draining`, `Drained`, or
    /// `Disabled` (admin axis), and — critically for auto-rejoin safety — they
    /// must NOT clear a `TemporaryBypass` / `RecoveredPendingCanary` quarantine.
    /// Only the recovery probe/canary loop may do that. The observation routes
    /// through [`WorkerLifecycle::observe_health`], which refuses to overwrite a
    /// quarantine, so a single lucky health check can never auto-rejoin a
    /// bypassed worker. Returns the resulting legacy [`WorkerStatus`].
    pub async fn apply_health_status(&self, health_status: WorkerStatus) -> WorkerStatus {
        let mut lifecycle = self.lifecycle.write().await;
        if let Some(eligibility) = health_status_to_eligibility(health_status) {
            lifecycle.observe_health(eligibility);
        }
        lifecycle.legacy_status()
    }

    /// Quarantine this worker into [`EligibilityState::TemporaryBypass`] because
    /// a probe hit a concrete failure `class`. Never touches the admin axis
    /// (transient failure ≠ desired inventory). The worker immediately drops out
    /// of scheduling — its [`Self::status`] reads `Unreachable` — and stays out
    /// until the recovery probe/canary loop clears it.
    pub async fn enter_bypass(&self, class: BypassFailureClass) {
        self.lifecycle.write().await.enter_bypass(class);
    }

    /// Advance a bypassed worker to [`EligibilityState::RecoveredPendingCanary`]
    /// after a fully-healthy recovery-probe streak. Legal only from
    /// `TemporaryBypass`.
    pub async fn recover_to_canary(&self) -> Result<(), IllegalLifecycleTransition> {
        self.lifecycle.write().await.recover_to_canary()
    }

    /// Fully rejoin a worker after a passing canary build. Legal only from
    /// `RecoveredPendingCanary`; never changes admin intent (an
    /// operator-disabled worker stays disabled even after a clean canary).
    pub async fn promote_from_canary(&self) -> Result<(), IllegalLifecycleTransition> {
        self.lifecycle.write().await.promote_from_canary()
    }

    /// Whether this worker is cleared for exactly one canary build.
    pub async fn is_canary_pending(&self) -> bool {
        self.lifecycle.read().await.is_canary_pending()
    }

    /// Configured ceiling reduced by current disk headroom.
    pub async fn effective_total_slots(&self) -> u32 {
        let config = self.config.read().await;
        let pressure = self.pressure_assessment.read().await;
        self.disk_slot_policy
            .effective_slots(config.total_slots, &pressure)
    }

    /// Available slots after disk derating and existing reservations.
    pub async fn available_slots(&self) -> u32 {
        let total = self.effective_total_slots().await;
        let used = self.used_slots.load(Ordering::Relaxed);
        total.saturating_sub(used)
    }

    /// Continuous disk-headroom credit using the same floor as slot admission.
    pub async fn disk_headroom(&self) -> f64 {
        let assessment = self.pressure_assessment.read().await;
        self.disk_slot_policy.headroom(&assessment)
    }

    /// Reserve slots for a job. Returns true if successful.
    ///
    /// Re-reads disk capacity on each CAS iteration and keeps the config and
    /// pressure snapshot locked through the reservation to avoid overallocation.
    ///
    /// Also re-checks status on each iteration and refuses reservations on
    /// workers that are `Draining`, `Drained`, or `Disabled`. The selector
    /// filters by `healthy_workers()` up-front, but between that filter and
    /// this CAS an operator may run `rch workers drain`, which would
    /// otherwise let a new build land on a worker that was just asked to
    /// stop accepting work — defeating the entire drain contract.
    pub async fn reserve_slots(&self, count: u32) -> bool {
        if count == 0 {
            return false;
        }

        let mut current = self.used_slots.load(Ordering::Relaxed);
        loop {
            // Re-read the lifecycle on each iteration so a concurrent `drain()`
            // / `disable()` (admin axis) or a freshly-entered `TemporaryBypass`
            // / `RecoveredPendingCanary` quarantine (eligibility axis) becomes
            // authoritative as soon as it's written, not only on the next
            // selection round.
            if !lifecycle_accepts_new_builds(*self.lifecycle.read().await) {
                return false;
            }
            let config = self.config.read().await;
            let pressure = self.pressure_assessment.read().await;
            let total_slots = self
                .disk_slot_policy
                .effective_slots(config.total_slots, &pressure);
            let Some(next) = current.checked_add(count) else {
                return false;
            };
            if next > total_slots {
                return false;
            }
            let reserved = self.used_slots.compare_exchange(
                current,
                next,
                Ordering::SeqCst,
                Ordering::Relaxed,
            );
            drop(pressure);
            drop(config);
            match reserved {
                Ok(_) => {
                    // TOCTOU defense: re-check the lifecycle after a successful
                    // slot reservation. If a concurrent drain()/disable() or a
                    // bypass quarantine landed between our initial read and the
                    // CAS, roll the reservation back.
                    if lifecycle_accepts_new_builds(*self.lifecycle.read().await)
                        && self.used_slots() <= self.effective_total_slots().await
                    {
                        return true;
                    }
                    // Roll back the reservation; release_slots also handles
                    // transitioning Draining -> Drained if this was the last slot.
                    self.release_slots(count).await;
                    return false;
                }
                Err(actual) => current = actual,
            }
        }
    }

    /// Release slots after a job completes.
    ///
    /// If the worker is in `Draining` state and all slots are now free,
    /// automatically transitions to `Drained` state.
    pub async fn release_slots(&self, count: u32) {
        let mut current = self.used_slots.load(Ordering::Relaxed);
        loop {
            let new_val = current.saturating_sub(count);

            if count > current {
                let id = self.config.read().await.id.clone();
                tracing::warn!(
                    "Worker {}: attempted to release {} slots but only {} in use",
                    id,
                    count,
                    current
                );
            }

            match self.used_slots.compare_exchange(
                current,
                new_val,
                Ordering::SeqCst,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(actual) => current = actual,
            }
        }

        // Check if draining worker should transition to drained
        self.check_drain_complete().await;
    }

    /// Check if this worker has a cached copy of a project.
    pub async fn has_cached_project(&self, project: &str) -> bool {
        self.cached_projects
            .read()
            .await
            .iter()
            .any(|p| p == project)
    }

    /// Add a project to the cache list.
    pub async fn add_cached_project(&self, project: String) {
        let mut cache = self.cached_projects.write().await;
        if !cache.contains(&project) {
            // Limit cache size to prevent unbounded growth
            if cache.len() >= 100 {
                cache.remove(0);
            }
            cache.push(project);
        }
    }

    /// Update the speed score.
    pub fn set_speed_score(&self, score: f64) {
        self.speed_score.store(score.to_bits(), Ordering::Relaxed);
    }

    /// Get the current speed score.
    pub fn get_speed_score(&self) -> f64 {
        f64::from_bits(self.speed_score.load(Ordering::Relaxed))
    }

    /// Update the last observed worker latency (ms).
    pub fn set_last_latency_ms(&self, latency_ms: Option<u64>) {
        self.last_latency_ms
            .store(latency_ms.unwrap_or(0), Ordering::Relaxed);
    }

    /// Get the last observed worker latency (ms).
    pub fn last_latency_ms(&self) -> Option<u64> {
        let v = self.last_latency_ms.load(Ordering::Relaxed);
        if v == 0 { None } else { Some(v) }
    }

    /// Get the current circuit breaker state.
    pub async fn circuit_state(&self) -> Option<CircuitState> {
        Some(self.circuit.read().await.state())
    }

    /// Get the circuit stats (for internal use).
    pub async fn circuit_stats(&self) -> CircuitStats {
        self.circuit.read().await.clone()
    }

    /// Record a successful operation for circuit breaker.
    pub async fn record_success(&self) {
        let mut circuit = self.circuit.write().await;
        circuit.record_success();
    }

    /// Drive the authoritative circuit through a single health-check outcome.
    ///
    /// This is the ONE write path the health monitor uses to move
    /// `WorkerState.circuit` — the circuit the scheduler actually reads — so a
    /// worker that fails a probe short-circuits out of selection and a recovered
    /// worker rejoins, all through the shared
    /// [`CircuitStats::apply_health_outcome`] engine. Returns
    /// `(previous_state, new_state)` so the caller can detect transitions for
    /// alerting/metrics without a second lock.
    ///
    /// `transient` flags a retryable transport blip that survived the in-check
    /// retries; a transient half-open failure will not reopen the circuit.
    pub async fn record_health_check(
        &self,
        healthy: bool,
        transient: bool,
        config: &CircuitBreakerConfig,
    ) -> (CircuitState, CircuitState) {
        let mut circuit = self.circuit.write().await;
        let previous = circuit.state();
        let new = circuit.apply_health_outcome(healthy, transient, config);
        (previous, new)
    }

    /// Drive the authoritative circuit through a single *command-class*
    /// outcome (see [`CircuitStats::apply_command_outcome`]): an operation of
    /// the same shape as dispatched work — the telemetry poll, or a health
    /// probe whose session connected but whose command never completed.
    /// Returns `(previous_state, new_state)` like [`Self::record_health_check`].
    pub async fn record_command_outcome(
        &self,
        success: bool,
        config: &CircuitBreakerConfig,
    ) -> (CircuitState, CircuitState) {
        let mut circuit = self.circuit.write().await;
        let previous = circuit.state();
        let new = circuit.apply_command_outcome(success, config);
        (previous, new)
    }

    /// Record a failed operation for circuit breaker.
    pub async fn record_failure(&self, error_msg: Option<String>) {
        let mut circuit = self.circuit.write().await;
        circuit.record_failure();

        if let Some(msg) = error_msg {
            *self.last_error_msg.write().await = Some(msg);
        }
    }

    /// Check if circuit should open based on config.
    pub async fn should_open_circuit(&self, config: &CircuitBreakerConfig) -> bool {
        self.circuit.read().await.should_open(config)
    }

    /// Check if circuit should transition to half-open.
    pub async fn should_half_open(&self, config: &CircuitBreakerConfig) -> bool {
        self.circuit.read().await.should_half_open(config)
    }

    /// Check if circuit should close.
    pub async fn should_close_circuit(&self, config: &CircuitBreakerConfig) -> bool {
        self.circuit.read().await.should_close(config)
    }

    /// Check if we can send a probe request.
    pub async fn can_probe(&self, config: &CircuitBreakerConfig) -> bool {
        self.circuit.read().await.can_probe(config)
    }

    /// Start a probe request.
    pub async fn start_probe(&self, config: &CircuitBreakerConfig) -> bool {
        self.circuit.write().await.start_probe(config)
    }

    /// Open the circuit.
    pub async fn open_circuit(&self) {
        self.circuit.write().await.open();
    }

    /// Transition to half-open.
    pub async fn half_open_circuit(&self) {
        self.circuit.write().await.half_open();
    }

    /// Close the circuit.
    pub async fn close_circuit(&self) {
        self.circuit.write().await.close();
        // Clear last error on circuit close
        *self.last_error_msg.write().await = None;
    }

    /// Get the last error message.
    pub async fn last_error(&self) -> Option<String> {
        self.last_error_msg.read().await.clone()
    }

    /// Set an error message.
    pub async fn set_error(&self, msg: String) {
        *self.last_error_msg.write().await = Some(msg);
    }

    /// Skip overlapping probes instead of queuing more remote inventory scans.
    /// The guard releases on normal completion, error, panic or cancellation.
    pub fn try_capability_probe(&self) -> Option<tokio::sync::MutexGuard<'_, ()>> {
        self.capability_probe.try_lock().ok()
    }

    /// Update worker capabilities.
    pub async fn set_capabilities(&self, capabilities: WorkerCapabilities) {
        let pressure_config = DiskPressurePolicyConfig::default();
        let pressure = evaluate_pressure_policy(&capabilities, None, &pressure_config);
        *self.capabilities.write().await = capabilities;

        let now_ms = current_unix_ms();
        let mut current = self.pressure_assessment.write().await;
        if pressure.state != crate::disk_pressure::PressureState::Critical
            && refresh_preserved_pressure_age(&mut current, &pressure_config, now_ms)
        {
            current.disk_free_gb = pressure.disk_free_gb;
            current.disk_total_gb = pressure.disk_total_gb;
            current.disk_free_ratio = pressure.disk_free_ratio;
            current.evaluated_at_unix_ms = now_ms;
            return;
        }

        *current = pressure;
    }

    /// Get worker capabilities.
    pub async fn capabilities(&self) -> WorkerCapabilities {
        self.capabilities.read().await.clone()
    }

    /// Cache a toolchain preflight verdict for selection-time routing.
    pub async fn record_toolchain_preflight(
        &self,
        toolchain: String,
        usable: bool,
        reason: Option<String>,
    ) {
        self.toolchain_preflight.write().await.insert(
            toolchain,
            ToolchainPreflightStatus {
                usable,
                reason,
                checked_at_unix_ms: current_unix_ms(),
            },
        );
    }

    /// Retrieve a cached toolchain preflight verdict.
    pub async fn toolchain_preflight_status(
        &self,
        toolchain: &str,
    ) -> Option<ToolchainPreflightStatus> {
        self.toolchain_preflight
            .read()
            .await
            .get(toolchain)
            .cloned()
    }

    /// Update the latest disk-pressure policy assessment.
    pub async fn set_pressure_assessment(&self, assessment: PressureAssessment) {
        *self.pressure_assessment.write().await = assessment;
    }

    /// Get the latest disk-pressure policy assessment.
    pub async fn pressure_assessment(&self) -> PressureAssessment {
        self.pressure_assessment.read().await.clone()
    }

    /// Check if this worker has Bun installed.
    pub async fn has_bun(&self) -> bool {
        self.capabilities.read().await.has_bun()
    }

    /// Check if this worker has Node.js installed.
    pub async fn has_node(&self) -> bool {
        self.capabilities.read().await.has_node()
    }

    /// Check if this worker has Rust installed.
    pub async fn has_rust(&self) -> bool {
        self.capabilities.read().await.has_rust()
    }

    /// Check if this worker has Nix installed (nix binary + `/nix/store`).
    pub async fn has_nix(&self) -> bool {
        self.capabilities.read().await.has_nix()
    }

    /// Whether this worker has the Go toolchain installed.
    pub async fn has_go(&self) -> bool {
        self.capabilities.read().await.has_go()
    }

    /// Whether this worker can run `cargo zigbuild` (cargo-zigbuild + zig).
    pub async fn has_zig(&self) -> bool {
        self.capabilities.read().await.has_zig()
    }

    /// Disable the worker with an optional reason.
    /// Sets status to Disabled and records the timestamp and reason.
    pub async fn disable(&self, reason: Option<String>) {
        *self.drain_completion.write().await = DrainCompletionAction::StayDrained;
        self.lifecycle
            .write()
            .await
            .set_admin(AdminIntent::Disabled);
        *self.disabled_reason.write().await = reason;
        self.disabled_at
            .store(current_unix_secs_for_disabled_at(), Ordering::Relaxed);
    }

    /// Start draining the worker (no new jobs, but finish existing).
    pub async fn drain(&self) {
        *self.drain_completion.write().await = DrainCompletionAction::StayDrained;
        self.lifecycle
            .write()
            .await
            .set_admin(AdminIntent::Draining);
        self.check_drain_complete().await;
    }

    /// Start draining a worker that should be removed once active jobs finish.
    pub async fn drain_for_removal(&self) {
        *self.drain_completion.write().await = DrainCompletionAction::RemoveFromPool;
        self.lifecycle
            .write()
            .await
            .set_admin(AdminIntent::Draining);
        self.check_drain_complete().await;
    }

    /// Start draining a worker that should become disabled once active jobs finish.
    pub async fn drain_then_disable(&self, reason: Option<String>) {
        *self.drain_completion.write().await = DrainCompletionAction::Disable { reason };
        self.lifecycle
            .write()
            .await
            .set_admin(AdminIntent::Draining);
        self.check_drain_complete().await;
    }

    /// Enable a previously disabled/draining worker.
    /// Clears disabled reason and timestamp, sets status to Healthy.
    pub async fn enable(&self) {
        *self.drain_completion.write().await = DrainCompletionAction::StayDrained;
        // Operator re-enable is an authoritative override: restore both axes to
        // fully in-service (Active + Healthy) and clear any bypass cause.
        *self.lifecycle.write().await = WorkerLifecycle::new();
        *self.disabled_reason.write().await = None;
        self.disabled_at.store(0, Ordering::Relaxed);
        // Also reset the circuit breaker — `WorkerState.circuit` is the single
        // source of truth the scheduler reads, and re-enabling a worker whose
        // circuit is still Open/HalfOpen would leave it short-circuited out of
        // selection despite the operator explicitly bringing it back. This is a
        // REAL reset (clears failure/success counters and error history), not a
        // status flip. (Also clears last_error_msg, matching close_circuit.)
        self.circuit.write().await.close();
        *self.last_error_msg.write().await = None;
    }

    /// Check if worker is disabled (operator admin intent).
    pub async fn is_disabled(&self) -> bool {
        self.lifecycle.read().await.admin == AdminIntent::Disabled
    }

    /// Check if worker is draining (operator admin intent).
    pub async fn is_draining(&self) -> bool {
        self.lifecycle.read().await.admin == AdminIntent::Draining
    }

    /// Check if worker is drained (drain complete, no active jobs).
    pub async fn is_drained(&self) -> bool {
        self.lifecycle.read().await.admin == AdminIntent::Drained
    }

    /// Check if draining worker has completed all jobs and apply the recorded completion action.
    ///
    /// Plain drains transition to `Drained`, config-removal drains become pruneable, and
    /// drain-before-disable transitions to `Disabled` while preserving the disable reason.
    pub async fn check_drain_complete(&self) {
        let used = self.used_slots.load(Ordering::Acquire);
        if used != 0 || self.lifecycle.read().await.admin != AdminIntent::Draining {
            return;
        }

        let completion = self.drain_completion.read().await.clone();
        let mut lifecycle = self.lifecycle.write().await;
        if lifecycle.admin != AdminIntent::Draining || self.used_slots.load(Ordering::Acquire) != 0
        {
            return;
        }

        let disable_reason = match completion {
            DrainCompletionAction::StayDrained | DrainCompletionAction::RemoveFromPool => {
                lifecycle.set_admin(AdminIntent::Drained);
                None
            }
            DrainCompletionAction::Disable { reason } => {
                lifecycle.set_admin(AdminIntent::Disabled);
                Some(reason)
            }
        };
        drop(lifecycle);

        if let Some(reason) = disable_reason {
            *self.disabled_reason.write().await = reason;
            self.disabled_at
                .store(current_unix_secs_for_disabled_at(), Ordering::Relaxed);
            *self.drain_completion.write().await = DrainCompletionAction::StayDrained;
        }
    }

    async fn remove_after_drain_pending(&self) -> bool {
        *self.drain_completion.read().await == DrainCompletionAction::RemoveFromPool
    }

    /// Get the reason this worker was disabled (if any).
    pub async fn disabled_reason(&self) -> Option<String> {
        self.disabled_reason.read().await.clone()
    }

    /// Get the timestamp when this worker was disabled (Unix seconds).
    pub fn disabled_at(&self) -> Option<i64> {
        let v = self.disabled_at.load(Ordering::Relaxed);
        if v == 0 { None } else { Some(v) }
    }

    /// Get the number of slots currently in use.
    pub fn used_slots(&self) -> u32 {
        self.used_slots.load(Ordering::Relaxed)
    }

    /// Startup-only reconstruction, independent of present admission limits.
    pub fn restore_slots(&self, count: u32) -> std::io::Result<()> {
        self.used_slots
            .try_update(Ordering::SeqCst, Ordering::SeqCst, |used| {
                used.checked_add(count)
            })
            .map(|_| ())
            .map_err(|_| {
                std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "durable reservation overflow",
                )
            })
    }
}

/// Cached result of checking a concrete Rust toolchain on a worker.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ToolchainPreflightStatus {
    /// Whether both `rustc` and `cargo` work through `rustup run <toolchain>`.
    pub usable: bool,
    /// Stable diagnostic reason when unusable.
    pub reason: Option<String>,
    /// Epoch milliseconds when the verdict was recorded.
    pub checked_at_unix_ms: i64,
}

impl ToolchainPreflightStatus {
    /// Whether this verdict is fresh enough to reuse for dispatch decisions.
    pub fn is_fresh(&self, ttl: Duration) -> bool {
        let age_ms = current_unix_ms().saturating_sub(self.checked_at_unix_ms);
        age_ms <= duration_millis_i64(ttl)
    }

    /// Whether this verdict says anything about the toolchain itself.
    ///
    /// Only `toolchain_preflight_command_failed` means the probe ran and the
    /// toolchain is missing or broken. A connect failure or timeout (the probe
    /// runs over SSH against possibly loaded workers) says nothing about the
    /// toolchain, and caching it as "unusable" for the full TTL excluded a
    /// healthy worker from every build for that toolchain for 10 minutes.
    pub fn is_definitive(&self) -> bool {
        self.usable
            || self
                .reason
                .as_deref()
                .is_some_and(|reason| reason.starts_with("toolchain_preflight_command_failed"))
    }

    /// How long to reuse this verdict: `definitive_ttl` for a real answer,
    /// `transient_ttl` for a probe that never reached a conclusion.
    pub fn is_reusable(&self, definitive_ttl: Duration, transient_ttl: Duration) -> bool {
        self.is_fresh(if self.is_definitive() {
            definitive_ttl
        } else {
            transient_ttl
        })
    }
}

/// Pool of all workers.
#[derive(Clone)]
pub struct WorkerPool {
    workers: Arc<RwLock<HashMap<WorkerId, Arc<WorkerState>>>>,
    /// Track worker count atomically for sync access.
    worker_count: Arc<AtomicUsize>,
    disk_slot_policy: DiskSlotPolicy,
}

impl WorkerPool {
    /// Create a new empty worker pool.
    pub fn new() -> Self {
        Self::with_selection_config(&rch_common::SelectionConfig::default())
    }

    /// Create a pool whose current and subsequently reloaded workers share a disk budget.
    pub fn with_selection_config(config: &rch_common::SelectionConfig) -> Self {
        Self {
            workers: Arc::new(RwLock::new(HashMap::new())),
            worker_count: Arc::new(AtomicUsize::new(0)),
            disk_slot_policy: DiskSlotPolicy::from(config),
        }
    }

    /// Add a worker to the pool.
    pub async fn add_worker(&self, config: WorkerConfig) {
        let id = config.id.clone();

        {
            let workers = self.workers.read().await;
            if let Some(existing) = workers.get(&id) {
                debug!("Updating existing worker: {}", id);
                existing.update_config(config).await;
                return;
            }
        }

        let state = Arc::new(WorkerState::with_disk_slot_policy(
            config,
            self.disk_slot_policy,
        ));
        let mut workers = self.workers.write().await;
        // Check again under write lock
        if let Some(existing) = workers.get(&id) {
            // Race condition: added between read and write lock
            // Just update config on the existing one, drop the new state
            let config = state.config.read().await.clone();
            existing.update_config(config).await;
        } else {
            workers.insert(id.clone(), state);
            self.worker_count.fetch_add(1, Ordering::SeqCst);
            debug!("Added worker: {}", id);
        }
    }

    /// Add a worker with a pre-configured state (for testing).
    #[cfg(test)]
    pub async fn add_worker_state(&self, state: WorkerState) {
        let id = state.config.read().await.id.clone();
        let mut workers = self.workers.write().await;
        if workers.insert(id.clone(), Arc::new(state)).is_none() {
            self.worker_count.fetch_add(1, Ordering::SeqCst);
        }
        debug!("Added worker: {}", id);
    }

    /// Remove a worker from the pool.
    pub async fn remove_worker(&self, id: &WorkerId) -> bool {
        let mut workers = self.workers.write().await;
        if workers.remove(id).is_some() {
            self.worker_count.fetch_sub(1, Ordering::SeqCst);
            debug!("Removed worker: {}", id);
            true
        } else {
            false
        }
    }

    /// Prune workers that were removed from config and have finished draining.
    ///
    /// Returns the number of workers removed.
    pub async fn prune_drained(&self) -> usize {
        let snapshot: Vec<(WorkerId, Arc<WorkerState>)> = {
            let workers = self.workers.read().await;
            workers
                .iter()
                .map(|(id, worker)| (id.clone(), Arc::clone(worker)))
                .collect()
        };

        let mut to_remove = Vec::new();
        for (id, worker) in snapshot {
            worker.check_drain_complete().await;
            let status = worker.status().await;

            if status == WorkerStatus::Drained
                && worker.used_slots() == 0
                && worker.remove_after_drain_pending().await
            {
                to_remove.push((id, worker));
            }
        }

        let mut count = 0;
        let mut workers = self.workers.write().await;
        for (id, worker) in to_remove {
            let still_same_worker = workers
                .get(&id)
                .is_some_and(|current| Arc::ptr_eq(current, &worker));
            if !still_same_worker {
                continue;
            }

            worker.check_drain_complete().await;
            let should_remove = worker.status().await == WorkerStatus::Drained
                && worker.used_slots() == 0
                && worker.remove_after_drain_pending().await;
            if should_remove && workers.remove(&id).is_some() {
                count += 1;
                debug!("Pruned drained worker: {}", id);
            }
        }

        if count > 0 {
            self.worker_count.fetch_sub(count, Ordering::SeqCst);
        }

        count
    }

    /// Get the number of workers in the pool.
    pub fn len(&self) -> usize {
        self.worker_count.load(Ordering::SeqCst)
    }

    /// Check if pool is empty.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Get all workers (regardless of status) for health monitoring.
    pub async fn all_workers(&self) -> Vec<Arc<WorkerState>> {
        let workers = self.workers.read().await;
        workers.values().cloned().collect()
    }

    /// Get all healthy workers (for job assignment).
    ///
    /// Includes workers with status Healthy or Degraded.
    pub async fn healthy_workers(&self) -> Vec<Arc<WorkerState>> {
        let workers = self.workers.read().await;
        let mut healthy = Vec::new();
        for worker in workers.values() {
            let status = worker.status().await;
            if status == WorkerStatus::Healthy || status == WorkerStatus::Degraded {
                healthy.push(worker.clone());
            }
        }
        healthy
    }

    /// Get a worker by ID.
    pub async fn get(&self, id: &WorkerId) -> Option<Arc<WorkerState>> {
        let workers = self.workers.read().await;
        workers.get(id).cloned()
    }

    /// Update worker status.
    pub async fn set_status(&self, id: &WorkerId, status: WorkerStatus) {
        let workers = self.workers.read().await;
        if let Some(worker) = workers.get(id) {
            worker.set_status(status).await;
            debug!("Set {} status to {:?}", id, status);
        }
    }

    /// Release reserved slots on a worker.
    pub async fn release_slots(&self, id: &WorkerId, slots: u32) {
        let workers = self.workers.read().await;
        if let Some(worker) = workers.get(id) {
            worker.release_slots(slots).await;
            debug!("Released {} slots on worker {}", slots, id);
        } else {
            debug!("Worker {} not found, cannot release slots", id);
        }
    }
}

fn current_unix_secs_for_disabled_at() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(duration_secs_i64)
        .unwrap_or(1)
        .max(1)
}

impl Default for WorkerPool {
    fn default() -> Self {
        Self::new()
    }
}

fn current_unix_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(duration_millis_i64)
        .unwrap_or_default()
}

fn refresh_preserved_pressure_age(
    assessment: &mut PressureAssessment,
    config: &DiskPressurePolicyConfig,
    now_ms: i64,
) -> bool {
    if !assessment.telemetry_fresh {
        return false;
    }

    let Some(previous_age_secs) = assessment.telemetry_age_secs else {
        return false;
    };

    let elapsed_secs = now_ms
        .saturating_sub(assessment.evaluated_at_unix_ms)
        .max(0) as u64
        / 1_000;
    let age_secs = previous_age_secs.saturating_add(elapsed_secs);
    assessment.telemetry_age_secs = Some(age_secs);
    assessment.telemetry_fresh = age_secs <= config.telemetry_stale_after.as_secs();
    assessment.telemetry_fresh
}

// ============================================================================
// Worker API Response Types
// ============================================================================

/// Worker capabilities information.
#[derive(Debug, Serialize)]
pub struct WorkerCapabilitiesInfo {
    pub id: String,
    pub host: String,
    pub user: String,
    pub capabilities: WorkerCapabilities,
    pub pressure_assessment: PressureAssessment,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub refresh: Option<WorkerCapabilitiesRefreshInfo>,
}

/// Freshness information for a capabilities refresh request.
#[derive(Debug, Clone, Serialize)]
pub struct WorkerCapabilitiesRefreshInfo {
    pub attempted: bool,
    pub live: bool,
    pub source: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
}

impl WorkerCapabilitiesRefreshInfo {
    fn live_probe() -> Self {
        Self {
            attempted: true,
            live: true,
            source: "live_probe".to_string(),
            message: None,
        }
    }

    fn cached_after_probe_failure(message: impl Into<String>) -> Self {
        Self {
            attempted: true,
            live: false,
            source: "cached_after_probe_failure".to_string(),
            message: Some(message.into()),
        }
    }
}

/// Worker capabilities response.
#[derive(Debug, Serialize)]
pub struct WorkerCapabilitiesResponse {
    pub workers: Vec<WorkerCapabilitiesInfo>,
}

/// Response for worker state changes (drain/enable/disable).
#[derive(Debug, Serialize)]
pub struct WorkerStateResponse {
    pub status: String,
    pub worker_id: String,
    pub action: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub new_status: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub message: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub active_slots: Option<u32>,
}

// ============================================================================
// Worker API Handlers
// ============================================================================

/// Collect worker capabilities, optionally refreshing via SSH probe.
pub async fn get_workers_capabilities(
    ctx: &DaemonContext,
    refresh: bool,
) -> WorkerCapabilitiesResponse {
    let workers = ctx.pool.all_workers().await;
    let mut entries = Vec::with_capacity(workers.len());
    let mut refresh_results = if refresh {
        refresh_worker_capabilities_concurrently(&workers).await
    } else {
        HashMap::new()
    };

    for worker in workers {
        let (id, host, user) = {
            let config = worker.config.read().await;
            (
                config.id.to_string(),
                config.host.clone(),
                config.user.clone(),
            )
        };
        let capabilities = worker.capabilities().await;
        let pressure_assessment = worker.pressure_assessment().await;
        let refresh = refresh_results.remove(&id);

        entries.push(WorkerCapabilitiesInfo {
            id,
            host,
            user,
            capabilities,
            pressure_assessment,
            refresh,
        });
    }

    WorkerCapabilitiesResponse { workers: entries }
}

async fn refresh_worker_capabilities_concurrently(
    workers: &[Arc<WorkerState>],
) -> HashMap<String, WorkerCapabilitiesRefreshInfo> {
    let mut handles = Vec::with_capacity(workers.len());

    for worker in workers {
        let id = {
            let config = worker.config.read().await;
            config.id.to_string()
        };
        let worker = Arc::clone(worker);
        handles.push((
            id,
            tokio::spawn(async move { refresh_worker_capabilities_for_worker(worker).await }),
        ));
    }

    let mut results = HashMap::with_capacity(handles.len());
    for (id, handle) in handles {
        let refresh = match handle.await {
            Ok(refresh) => refresh,
            Err(err) => WorkerCapabilitiesRefreshInfo::cached_after_probe_failure(format!(
                "capabilities refresh task failed: {err}; returning cached capability snapshot"
            )),
        };
        results.insert(id, refresh);
    }

    results
}

async fn refresh_worker_capabilities_for_worker(
    worker: Arc<WorkerState>,
) -> WorkerCapabilitiesRefreshInfo {
    let probe = tokio::time::timeout(
        CAPABILITIES_REFRESH_WORKER_BUDGET,
        // On-demand capability refresh (the `rch workers capabilities` API path)
        // uses a throwaway SSH session — it is a low-frequency operator query, not
        // one of the ~30s periodic poll loops the shared pool exists to de-flood.
        probe_worker_capabilities(&worker, CAPABILITIES_REFRESH_PROBE_TIMEOUT, None),
    )
    .await;

    match probe {
        Ok(Some(capabilities)) => {
            worker.set_capabilities(capabilities).await;
            WorkerCapabilitiesRefreshInfo::live_probe()
        }
        Ok(None) => WorkerCapabilitiesRefreshInfo::cached_after_probe_failure(
            "capabilities probe failed; returning cached capability snapshot",
        ),
        Err(_) => WorkerCapabilitiesRefreshInfo::cached_after_probe_failure(format!(
            "capabilities probe exceeded {}s response budget; returning cached capability snapshot",
            CAPABILITIES_REFRESH_WORKER_BUDGET.as_secs()
        )),
    }
}

/// Handle a worker drain request.
pub async fn handle_worker_drain(ctx: &DaemonContext, worker_id: &WorkerId) -> WorkerStateResponse {
    match ctx.pool.get(worker_id).await {
        Some(worker) => {
            let active_slots = worker.used_slots();
            worker.drain().await;
            let new_status = worker.status().await;
            WorkerStateResponse {
                status: "ok".to_string(),
                worker_id: worker_id.to_string(),
                action: "drain".to_string(),
                new_status: Some(worker_status_label(new_status).to_string()),
                reason: None,
                message: Some(format!(
                    "Worker is {}. {} slot(s) currently in use.",
                    worker_status_label(new_status),
                    active_slots,
                )),
                active_slots: Some(active_slots),
            }
        }
        None => WorkerStateResponse {
            status: "error".to_string(),
            worker_id: worker_id.to_string(),
            action: "drain".to_string(),
            new_status: None,
            reason: None,
            message: Some(format!("Worker '{}' not found", worker_id)),
            active_slots: None,
        },
    }
}

/// Handle a worker enable request.
pub async fn handle_worker_enable(
    ctx: &DaemonContext,
    worker_id: &WorkerId,
) -> WorkerStateResponse {
    match ctx.pool.get(worker_id).await {
        Some(worker) => {
            let old_status = worker.status().await;
            worker.enable().await;
            // Operator re-enable must be DURABLE: delete the worker's persisted
            // bypass record too. `worker.enable()` only resets the in-memory
            // lifecycle/circuit; if the record survives, `reconcile_on_start`
            // re-quarantines the worker from it on the next daemon restart (and
            // the live recovery loop can re-touch it). Mirrors `rejoin`'s
            // `store.remove(worker_id)`. (2026-07-16 offload meltdown follow-up.)
            if let Some(store) = &ctx.bypass_store {
                let _ = store.lock().await.remove(worker_id.as_str());
            }
            // Same durability rule for the admin-disable record (bd-8zxz7):
            // an enable that leaves the record behind would be undone by the
            // next daemon restart's re-apply pass.
            if let Some(store) = &ctx.admin_disable_store
                && let Err(e) = store.lock().await.remove(worker_id.as_str())
            {
                tracing::warn!(
                    "Failed to remove durable admin-disable record for {}: {}",
                    worker_id,
                    e
                );
            }
            WorkerStateResponse {
                status: "ok".to_string(),
                worker_id: worker_id.to_string(),
                action: "enable".to_string(),
                new_status: Some("healthy".to_string()),
                reason: None,
                message: Some(format!("Worker enabled (was {:?})", old_status)),
                active_slots: None,
            }
        }
        None => WorkerStateResponse {
            status: "error".to_string(),
            worker_id: worker_id.to_string(),
            action: "enable".to_string(),
            new_status: None,
            reason: None,
            message: Some(format!("Worker '{}' not found", worker_id)),
            active_slots: None,
        },
    }
}

/// Handle a worker disable request.
pub async fn handle_worker_disable(
    ctx: &DaemonContext,
    worker_id: &WorkerId,
    reason: Option<String>,
    drain_first: bool,
) -> WorkerStateResponse {
    match ctx.pool.get(worker_id).await {
        Some(worker) => {
            let active_slots = worker.used_slots();

            // Durable admin disable (bd-8zxz7): persist the intent BEFORE the
            // in-memory flip, for both the drain and immediate branches — a
            // disable that only lives in memory evaporates on daemon restart
            // (observed live: the ovh-b cpu-capability quarantine rejoined the
            // pool after a kickstart). Persist failure is logged but does not
            // block the in-memory disable (fail-open, matches bypass_store).
            if let Some(store) = &ctx.admin_disable_store {
                let record = rch_common::bypass_record::AdminDisableRecord {
                    worker_id: worker_id.to_string(),
                    reason: reason.clone(),
                    disabled_unix_ms: u64::try_from(chrono::Utc::now().timestamp_millis())
                        .unwrap_or(0),
                };
                if let Err(e) = store.lock().await.upsert(record) {
                    tracing::warn!(
                        "Failed to persist durable admin-disable record for {}: {}",
                        worker_id,
                        e
                    );
                }
            }

            if drain_first && active_slots > 0 {
                worker.drain_then_disable(reason.clone()).await;
                let new_status = worker.status().await;
                WorkerStateResponse {
                    status: "ok".to_string(),
                    worker_id: worker_id.to_string(),
                    action: "disable".to_string(),
                    new_status: Some(worker_status_label(new_status).to_string()),
                    reason: reason.clone(),
                    message: Some(format!(
                        "Worker is {} before disable. {} slot(s) in use. Reason: {}",
                        worker_status_label(new_status),
                        active_slots,
                        reason.as_deref().unwrap_or("none provided")
                    )),
                    active_slots: Some(active_slots),
                }
            } else {
                // Disable immediately
                worker.disable(reason.clone()).await;
                WorkerStateResponse {
                    status: "ok".to_string(),
                    worker_id: worker_id.to_string(),
                    action: "disable".to_string(),
                    new_status: Some("disabled".to_string()),
                    reason: reason.clone(),
                    message: Some(format!(
                        "Worker disabled. Reason: {}",
                        reason.as_deref().unwrap_or("none provided")
                    )),
                    active_slots: Some(active_slots),
                }
            }
        }
        None => WorkerStateResponse {
            status: "error".to_string(),
            worker_id: worker_id.to_string(),
            action: "disable".to_string(),
            new_status: None,
            reason,
            message: Some(format!("Worker '{}' not found", worker_id)),
            active_slots: None,
        },
    }
}

fn worker_status_label(status: WorkerStatus) -> &'static str {
    match status {
        WorkerStatus::Healthy => "healthy",
        WorkerStatus::Degraded => "degraded",
        WorkerStatus::Unreachable => "unreachable",
        WorkerStatus::Draining => "draining",
        WorkerStatus::Drained => "drained",
        WorkerStatus::Disabled => "disabled",
    }
}

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

    // =========================================================================
    // Worker lifecycle type model (bd-session-history-remediation-ocv9i.1.1)
    // =========================================================================

    /// Every concrete failure class, for exhaustive bypass-attribution tests.
    const ALL_FAILURE_CLASSES: &[BypassFailureClass] = &[
        BypassFailureClass::Ssh,
        BypassFailureClass::WorkerBinary,
        BypassFailureClass::RuntimeToolchain,
        BypassFailureClass::DiskInodePressure,
        BypassFailureClass::StaleTelemetry,
        BypassFailureClass::PathSync,
        BypassFailureClass::ArtifactRetrieval,
        BypassFailureClass::CircuitBreaker,
        BypassFailureClass::OsArchMismatch,
    ];

    #[test]
    fn lifecycle_new_is_active_healthy_and_schedulable() {
        let lc = WorkerLifecycle::new();
        assert_eq!(lc.admin, AdminIntent::Active);
        assert_eq!(lc.eligibility, EligibilityState::Healthy);
        assert_eq!(lc.bypass_cause, None);
        assert!(lc.is_schedulable());
        assert!(!lc.is_canary_pending());
        // Default must match new().
        assert_eq!(WorkerLifecycle::default(), lc);
    }

    #[test]
    fn lifecycle_from_worker_status_maps_both_axes() {
        let cases = [
            (
                WorkerStatus::Healthy,
                AdminIntent::Active,
                EligibilityState::Healthy,
            ),
            (
                WorkerStatus::Degraded,
                AdminIntent::Active,
                EligibilityState::Degraded,
            ),
            (
                WorkerStatus::Unreachable,
                AdminIntent::Active,
                EligibilityState::Unreachable,
            ),
            (
                WorkerStatus::Draining,
                AdminIntent::Draining,
                EligibilityState::Healthy,
            ),
            (
                WorkerStatus::Drained,
                AdminIntent::Drained,
                EligibilityState::Healthy,
            ),
            (
                WorkerStatus::Disabled,
                AdminIntent::Disabled,
                EligibilityState::Healthy,
            ),
        ];
        for (status, admin, eligibility) in cases {
            let lc = WorkerLifecycle::from_worker_status(status);
            assert_eq!(lc.admin, admin, "admin axis for {status:?}");
            assert_eq!(
                lc.eligibility, eligibility,
                "eligibility axis for {status:?}"
            );
        }
    }

    #[test]
    fn lifecycle_legacy_status_collapses_two_axes() {
        // Admin intent dominates.
        let disabled = WorkerLifecycle {
            admin: AdminIntent::Disabled,
            eligibility: EligibilityState::Healthy,
            bypass_cause: None,
        };
        assert_eq!(disabled.legacy_status(), WorkerStatus::Disabled);
        // New transient states map to their closest legacy equivalent.
        let bypassed = WorkerLifecycle {
            admin: AdminIntent::Active,
            eligibility: EligibilityState::TemporaryBypass,
            bypass_cause: Some(BypassFailureClass::Ssh),
        };
        assert_eq!(bypassed.legacy_status(), WorkerStatus::Unreachable);
        let canary = WorkerLifecycle {
            admin: AdminIntent::Active,
            eligibility: EligibilityState::RecoveredPendingCanary,
            bypass_cause: None,
        };
        // Canary-pending must collapse to Unreachable (NOT Degraded), so the
        // legacy scheduler keeps it out of normal builds (one-canary-first).
        assert_eq!(canary.legacy_status(), WorkerStatus::Unreachable);
        // Round-trip the unambiguous health states.
        for status in [
            WorkerStatus::Healthy,
            WorkerStatus::Degraded,
            WorkerStatus::Disabled,
        ] {
            assert_eq!(
                WorkerLifecycle::from_worker_status(status).legacy_status(),
                status
            );
        }
    }

    #[test]
    fn lifecycle_is_schedulable_matrix() {
        // Active + Healthy/Degraded => schedulable.
        for elig in [EligibilityState::Healthy, EligibilityState::Degraded] {
            let lc = WorkerLifecycle {
                admin: AdminIntent::Active,
                eligibility: elig,
                bypass_cause: None,
            };
            assert!(lc.is_schedulable(), "Active+{elig:?} must schedule");
        }
        // Active + non-eligible => not schedulable.
        for elig in [
            EligibilityState::Unreachable,
            EligibilityState::TemporaryBypass,
            EligibilityState::RecoveredPendingCanary,
        ] {
            let lc = WorkerLifecycle {
                admin: AdminIntent::Active,
                eligibility: elig,
                bypass_cause: None,
            };
            assert!(!lc.is_schedulable(), "Active+{elig:?} must NOT schedule");
        }
        // Any non-Active admin intent => never schedulable, even when Healthy.
        for admin in [
            AdminIntent::Draining,
            AdminIntent::Drained,
            AdminIntent::Disabled,
        ] {
            let lc = WorkerLifecycle {
                admin,
                eligibility: EligibilityState::Healthy,
                bypass_cause: None,
            };
            assert!(!lc.is_schedulable(), "{admin:?}+Healthy must NOT schedule");
        }
    }

    #[test]
    fn lifecycle_enter_bypass_quarantines_for_every_failure_class_without_touching_admin() {
        for &cause in ALL_FAILURE_CLASSES {
            let mut lc = WorkerLifecycle::new();
            let admin_before = lc.admin;
            lc.enter_bypass(cause);
            assert_eq!(lc.eligibility, EligibilityState::TemporaryBypass);
            assert_eq!(lc.bypass_cause, Some(cause), "cause recorded for {cause:?}");
            // The cardinal invariant: transient failure never mutates desired
            // inventory.
            assert_eq!(lc.admin, admin_before, "admin axis must be untouched");
            assert!(!lc.is_schedulable());
        }
    }

    #[test]
    fn lifecycle_legal_recovery_path_healthy_again() {
        let mut lc = WorkerLifecycle::new();
        lc.enter_bypass(BypassFailureClass::DiskInodePressure);
        lc.recover_to_canary().expect("probe success -> canary");
        assert_eq!(lc.eligibility, EligibilityState::RecoveredPendingCanary);
        assert_eq!(lc.bypass_cause, None, "cause cleared on recovery");
        assert!(lc.is_canary_pending());
        assert!(
            !lc.is_schedulable(),
            "canary-pending is not normal scheduling"
        );
        lc.promote_from_canary().expect("canary success -> healthy");
        assert_eq!(lc.eligibility, EligibilityState::Healthy);
        assert!(lc.is_schedulable());
    }

    #[test]
    fn lifecycle_canary_relapse_via_enter_bypass_is_legal() {
        let mut lc = WorkerLifecycle::new();
        lc.enter_bypass(BypassFailureClass::Ssh);
        lc.recover_to_canary().unwrap();
        // Canary build fails -> relapse back to bypass with the new cause.
        lc.enter_bypass(BypassFailureClass::PathSync);
        assert_eq!(lc.eligibility, EligibilityState::TemporaryBypass);
        assert_eq!(lc.bypass_cause, Some(BypassFailureClass::PathSync));
    }

    #[test]
    fn lifecycle_illegal_recover_to_canary_from_non_bypass() {
        for elig in [
            EligibilityState::Healthy,
            EligibilityState::Degraded,
            EligibilityState::Unreachable,
            EligibilityState::RecoveredPendingCanary,
        ] {
            let mut lc = WorkerLifecycle {
                admin: AdminIntent::Active,
                eligibility: elig,
                bypass_cause: None,
            };
            let err = lc
                .recover_to_canary()
                .expect_err("recover_to_canary must be illegal from non-bypass");
            assert_eq!(err.from, elig);
            assert_eq!(err.to, EligibilityState::RecoveredPendingCanary);
            // State unchanged on illegal transition.
            assert_eq!(lc.eligibility, elig);
        }
    }

    #[test]
    fn lifecycle_illegal_promote_from_non_canary() {
        for elig in [
            EligibilityState::Healthy,
            EligibilityState::Degraded,
            EligibilityState::Unreachable,
            EligibilityState::TemporaryBypass,
        ] {
            let mut lc = WorkerLifecycle {
                admin: AdminIntent::Active,
                eligibility: elig,
                bypass_cause: None,
            };
            let err = lc
                .promote_from_canary()
                .expect_err("promote must be illegal from non-canary");
            assert_eq!(err.from, elig);
            assert_eq!(lc.eligibility, elig, "state unchanged on illegal promote");
        }
    }

    #[test]
    fn lifecycle_observe_health_never_clears_quarantine() {
        let mut lc = WorkerLifecycle::new();
        lc.enter_bypass(BypassFailureClass::StaleTelemetry);
        // A "healthy" heartbeat must NOT auto-clear a failure-class bypass.
        assert!(!lc.observe_health(EligibilityState::Healthy));
        assert_eq!(lc.eligibility, EligibilityState::TemporaryBypass);
        assert_eq!(lc.bypass_cause, Some(BypassFailureClass::StaleTelemetry));
        // Same protection while awaiting canary.
        lc.recover_to_canary().unwrap();
        assert!(!lc.observe_health(EligibilityState::Healthy));
        assert_eq!(lc.eligibility, EligibilityState::RecoveredPendingCanary);
    }

    #[test]
    fn lifecycle_observe_health_moves_among_plain_states() {
        let mut lc = WorkerLifecycle::new();
        assert!(lc.observe_health(EligibilityState::Degraded));
        assert_eq!(lc.eligibility, EligibilityState::Degraded);
        assert!(lc.observe_health(EligibilityState::Unreachable));
        assert_eq!(lc.eligibility, EligibilityState::Unreachable);
        assert!(lc.observe_health(EligibilityState::Healthy));
        assert_eq!(lc.eligibility, EligibilityState::Healthy);
        // Passing a non-health state is rejected without mutation.
        assert!(!lc.observe_health(EligibilityState::TemporaryBypass));
        assert_eq!(lc.eligibility, EligibilityState::Healthy);
    }

    #[test]
    fn lifecycle_operator_disabled_worker_never_auto_rejoins() {
        // A worker the operator disabled must stay out of service even if it
        // walks the full recovery path on the eligibility axis.
        let mut lc = WorkerLifecycle::new();
        lc.set_admin(AdminIntent::Disabled);
        lc.enter_bypass(BypassFailureClass::CircuitBreaker);
        lc.recover_to_canary().unwrap();
        lc.promote_from_canary().unwrap();
        assert_eq!(lc.eligibility, EligibilityState::Healthy);
        // Admin intent is the gate: still disabled => never schedulable.
        assert_eq!(lc.admin, AdminIntent::Disabled);
        assert!(!lc.is_schedulable());
        assert!(!lc.is_canary_pending());
    }

    #[test]
    fn lifecycle_admin_axis_is_invariant_across_eligibility_transitions() {
        for admin in [
            AdminIntent::Active,
            AdminIntent::Draining,
            AdminIntent::Drained,
            AdminIntent::Disabled,
        ] {
            let mut lc = WorkerLifecycle {
                admin,
                eligibility: EligibilityState::Healthy,
                bypass_cause: None,
            };
            lc.enter_bypass(BypassFailureClass::OsArchMismatch);
            lc.recover_to_canary().unwrap();
            lc.promote_from_canary().unwrap();
            lc.observe_health(EligibilityState::Degraded);
            assert_eq!(
                lc.admin, admin,
                "eligibility transitions must not move admin"
            );
        }
    }

    #[test]
    fn lifecycle_serde_round_trips_full_state() {
        let lc = WorkerLifecycle {
            admin: AdminIntent::Draining,
            eligibility: EligibilityState::TemporaryBypass,
            bypass_cause: Some(BypassFailureClass::ArtifactRetrieval),
        };
        let json = serde_json::to_string(&lc).expect("serialize");
        assert!(json.contains("temporary_bypass"));
        assert!(json.contains("artifact_retrieval"));
        let back: WorkerLifecycle = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(back, lc);
        // bypass_cause omitted when None.
        let healthy = WorkerLifecycle::new();
        let json = serde_json::to_string(&healthy).expect("serialize");
        assert!(!json.contains("bypass_cause"), "None cause must be omitted");
    }

    #[test]
    fn lifecycle_every_failure_class_serializes_to_distinct_snake_case_token() {
        let mut seen = std::collections::BTreeSet::new();
        for &cause in ALL_FAILURE_CLASSES {
            let token = serde_json::to_string(&cause).expect("serialize cause");
            assert!(seen.insert(token.clone()), "duplicate token {token}");
            // Round-trips.
            let back: BypassFailureClass = serde_json::from_str(&token).expect("deserialize cause");
            assert_eq!(back, cause);
        }
        assert_eq!(seen.len(), ALL_FAILURE_CLASSES.len());
    }

    #[test]
    fn test_duration_millis_i64_saturates() {
        assert_eq!(
            duration_millis_i64(Duration::from_millis(u64::MAX)),
            i64::MAX
        );
    }

    #[test]
    fn test_toolchain_preflight_freshness_saturates_extreme_ttl() {
        let status = ToolchainPreflightStatus {
            usable: true,
            reason: None,
            checked_at_unix_ms: 0,
        };

        assert!(status.is_fresh(Duration::from_millis(u64::MAX)));
    }

    /// Only a probe that ran and failed says the toolchain is broken; an
    /// unreachable or timed-out worker must not be excluded for the full TTL.
    #[test]
    fn test_toolchain_preflight_transport_failures_expire_quickly() {
        let long = Duration::from_secs(600);
        let short = Duration::from_secs(60);
        let two_minutes_ago = current_unix_ms() - 120_000;
        let verdict = |usable: bool, reason: Option<&str>| ToolchainPreflightStatus {
            usable,
            reason: reason.map(str::to_string),
            checked_at_unix_ms: two_minutes_ago,
        };

        let broken = verdict(
            false,
            Some("toolchain_preflight_command_failed:1:no such toolchain"),
        );
        assert!(broken.is_definitive());
        assert!(broken.is_reusable(long, short));

        let unreachable = verdict(false, Some("toolchain_preflight_connect_failed:timed out"));
        assert!(!unreachable.is_definitive());
        assert!(!unreachable.is_reusable(long, short));

        let healthy = verdict(true, None);
        assert!(healthy.is_definitive());
        assert!(healthy.is_reusable(long, short));
    }

    /// Helper to create a test worker config with given id.
    fn test_config(id: &str) -> WorkerConfig {
        WorkerConfig {
            id: WorkerId::new(id),
            host: "localhost".to_string(),
            user: "user".to_string(),
            identity_file: "~/.ssh/id_rsa".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        }
    }

    #[tokio::test]
    #[serial_test::serial]
    async fn test_refresh_worker_capabilities_marks_live_probe() {
        let _guard = rch_common::test_guard!();
        struct MockOverrideGuard;
        impl Drop for MockOverrideGuard {
            fn drop(&mut self) {
                rch_common::mock::clear_mock_overrides();
            }
        }

        rch_common::mock::set_mock_enabled_override(Some(true));
        let _mock_guard = MockOverrideGuard;
        let state = std::sync::Arc::new(WorkerState::new(test_config("test-worker")));

        let refresh = refresh_worker_capabilities_for_worker(std::sync::Arc::clone(&state)).await;

        assert!(refresh.attempted);
        assert!(refresh.live);
        assert_eq!(refresh.source, "live_probe");
        assert!(state.has_rust().await);
    }

    // ============== WorkerState Tests ==============

    #[tokio::test]
    async fn test_worker_state_new_defaults() {
        let state = WorkerState::new(test_config("test-worker"));

        // Check default values
        assert_eq!(state.status().await, WorkerStatus::Healthy);
        assert_eq!(state.available_slots().await, 8);
        assert_eq!(state.used_slots(), 0);
        assert_eq!(state.get_speed_score(), 50.0);
        assert!(state.last_latency_ms().is_none());
        assert!(state.last_error().await.is_none());
        assert!(!state.is_disabled().await);
        assert!(!state.is_draining().await);
        assert!(state.disabled_reason().await.is_none());
        assert!(state.disabled_at().is_none());
    }

    // =========================================================================
    // WS3.4: Unit tests for atomic field accessors (bd-1gyl)
    //
    // Verify that AtomicU64/AtomicI64 accessors for speed_score,
    // last_latency_ms, and disabled_at correctly roundtrip values including
    // edge cases (NaN, infinity, zero sentinels).
    // =========================================================================

    // --- speed_score (AtomicU64 via f64::to_bits/from_bits) ---

    #[test]
    fn test_speed_score_roundtrip_normal_values() {
        let state = WorkerState::new(test_config("test"));

        // Default should be 50.0
        assert_eq!(state.get_speed_score(), 50.0);

        // Set and get various normal values
        state.set_speed_score(0.0);
        assert_eq!(state.get_speed_score(), 0.0);

        state.set_speed_score(1.0);
        assert_eq!(state.get_speed_score(), 1.0);

        state.set_speed_score(100.0);
        assert_eq!(state.get_speed_score(), 100.0);

        state.set_speed_score(99.5);
        assert_eq!(state.get_speed_score(), 99.5);

        // Negative score (shouldn't happen in practice, but verify roundtrip)
        state.set_speed_score(-1.0);
        assert_eq!(state.get_speed_score(), -1.0);
    }

    #[test]
    fn test_speed_score_roundtrip_extremes() {
        let state = WorkerState::new(test_config("test"));

        // f64::MAX
        state.set_speed_score(f64::MAX);
        assert_eq!(state.get_speed_score(), f64::MAX);

        // f64::MIN (most negative)
        state.set_speed_score(f64::MIN);
        assert_eq!(state.get_speed_score(), f64::MIN);

        // f64::MIN_POSITIVE (smallest positive)
        state.set_speed_score(f64::MIN_POSITIVE);
        assert_eq!(state.get_speed_score(), f64::MIN_POSITIVE);

        // f64::EPSILON
        state.set_speed_score(f64::EPSILON);
        assert_eq!(state.get_speed_score(), f64::EPSILON);
    }

    #[test]
    fn test_speed_score_roundtrip_special_float_values() {
        let state = WorkerState::new(test_config("test"));

        // Positive infinity
        state.set_speed_score(f64::INFINITY);
        assert_eq!(state.get_speed_score(), f64::INFINITY);

        // Negative infinity
        state.set_speed_score(f64::NEG_INFINITY);
        assert_eq!(state.get_speed_score(), f64::NEG_INFINITY);

        // NaN roundtrips via to_bits/from_bits (NaN != NaN, so check bits)
        state.set_speed_score(f64::NAN);
        let retrieved = state.get_speed_score();
        assert!(
            retrieved.is_nan(),
            "NaN should roundtrip via to_bits/from_bits"
        );
        assert_eq!(
            f64::NAN.to_bits(),
            retrieved.to_bits(),
            "NaN bit pattern should be preserved"
        );

        // Negative zero
        state.set_speed_score(-0.0_f64);
        let retrieved = state.get_speed_score();
        assert_eq!(retrieved, 0.0); // -0.0 == 0.0 in f64
        // But bit pattern differs from +0.0
        assert_eq!((-0.0_f64).to_bits(), retrieved.to_bits());
    }

    // --- last_latency_ms (AtomicU64 with 0 as None sentinel) ---

    #[test]
    fn test_last_latency_ms_none_default() {
        let state = WorkerState::new(test_config("test"));
        assert_eq!(state.last_latency_ms(), None);
    }

    #[test]
    fn test_last_latency_ms_roundtrip() {
        let state = WorkerState::new(test_config("test"));

        // Set Some value
        state.set_last_latency_ms(Some(100));
        assert_eq!(state.last_latency_ms(), Some(100));

        // Update to different value
        state.set_last_latency_ms(Some(250));
        assert_eq!(state.last_latency_ms(), Some(250));

        // Large value
        state.set_last_latency_ms(Some(u64::MAX));
        assert_eq!(state.last_latency_ms(), Some(u64::MAX));

        // Set back to None
        state.set_last_latency_ms(None);
        assert_eq!(state.last_latency_ms(), None);
    }

    #[test]
    fn test_last_latency_ms_zero_sentinel_behavior() {
        let state = WorkerState::new(test_config("test"));

        // Some(0) maps to None because 0 is the sentinel value.
        // This is documented behavior: 0ms latency is not meaningful.
        state.set_last_latency_ms(Some(0));
        assert_eq!(
            state.last_latency_ms(),
            None,
            "Some(0) should map to None (0 is sentinel for None)"
        );

        // Some(1) should work fine
        state.set_last_latency_ms(Some(1));
        assert_eq!(state.last_latency_ms(), Some(1));
    }

    // --- disabled_at (AtomicI64 with 0 as None sentinel) ---

    #[test]
    fn test_disabled_at_none_default() {
        let state = WorkerState::new(test_config("test"));
        assert_eq!(state.disabled_at(), None);
    }

    #[tokio::test]
    async fn test_disabled_at_set_on_disable() {
        let state = WorkerState::new(test_config("test"));

        // Initially None
        assert!(state.disabled_at().is_none());

        // Disabling the worker should set disabled_at
        state.disable(Some("test reason".to_string())).await;
        let ts = state.disabled_at();
        assert!(ts.is_some(), "disabled_at should be set after disable()");
        assert!(ts.unwrap() > 0, "timestamp should be positive");
    }

    #[tokio::test]
    async fn test_disabled_at_cleared_on_enable() {
        let state = WorkerState::new(test_config("test"));

        // Disable then enable
        state.disable(Some("test".to_string())).await;
        assert!(state.disabled_at().is_some());

        state.enable().await;
        assert_eq!(
            state.disabled_at(),
            None,
            "disabled_at should be None after enable()"
        );
    }

    #[tokio::test]
    async fn test_record_health_check_drives_authoritative_circuit() {
        // The health monitor's ONLY circuit write path is record_health_check;
        // this is the circuit the scheduler reads via circuit_state(). Failures
        // must open it, and a healthy streak must recover it — all through the
        // shared apply_health_outcome engine.
        let state = WorkerState::new(test_config("test"));
        let cfg = CircuitBreakerConfig {
            failure_threshold: 3,
            success_threshold: 2,
            open_cooldown_secs: 0,
            ..Default::default()
        };

        // Starts Closed.
        assert_eq!(state.circuit_state().await, Some(CircuitState::Closed));

        // Three failures open the authoritative circuit.
        for _ in 0..3 {
            state.record_health_check(false, false, &cfg).await;
        }
        // With a 0s cooldown the engine immediately re-arms half-open after
        // opening; the point is the circuit is no longer Closed, so selection
        // sees it as non-healthy.
        assert_ne!(state.circuit_state().await, Some(CircuitState::Closed));

        // Two clean probes recover it to Closed.
        state.record_health_check(true, false, &cfg).await;
        let (_, new) = state.record_health_check(true, false, &cfg).await;
        assert_eq!(new, CircuitState::Closed);
        assert_eq!(state.circuit_state().await, Some(CircuitState::Closed));
    }

    #[tokio::test]
    async fn test_enable_resets_authoritative_circuit() {
        // enable() must REALLY reset the circuit the scheduler reads, not just
        // flip lifecycle status — otherwise a re-enabled worker stays
        // short-circuited out of selection.
        let state = WorkerState::new(test_config("test"));

        // Force the circuit Open, then record an error message.
        state.open_circuit().await;
        state.set_error("boom".to_string()).await;
        assert_eq!(state.circuit_state().await, Some(CircuitState::Open));

        // Operator re-enable must close the circuit and clear the error.
        state.enable().await;
        assert_eq!(
            state.circuit_state().await,
            Some(CircuitState::Closed),
            "enable() must reset the authoritative circuit to Closed"
        );
        assert_eq!(state.last_error().await, None);
    }

    // --- Concurrent access tests ---

    #[tokio::test]
    async fn test_speed_score_concurrent_updates() {
        let state = Arc::new(WorkerState::new(test_config("test")));
        let mut handles = vec![];

        // Spawn 10 tasks that each set a different speed score
        for i in 0..10u64 {
            let s = Arc::clone(&state);
            handles.push(tokio::spawn(async move {
                s.set_speed_score(i as f64 * 10.0);
            }));
        }

        for h in handles {
            h.await.unwrap();
        }

        // The final value should be one of the set values (last-write-wins)
        let score = state.get_speed_score();
        assert!(
            (0.0..=90.0).contains(&score) && score % 10.0 == 0.0,
            "Final speed score {} should be one of the written values",
            score
        );
    }

    #[tokio::test]
    async fn test_latency_concurrent_updates() {
        let state = Arc::new(WorkerState::new(test_config("test")));
        let mut handles = vec![];

        // Spawn tasks that set latency and None concurrently
        for i in 0..10u64 {
            let s = Arc::clone(&state);
            handles.push(tokio::spawn(async move {
                if i % 2 == 0 {
                    s.set_last_latency_ms(Some(i * 100 + 1)); // Avoid 0
                } else {
                    s.set_last_latency_ms(None);
                }
            }));
        }

        for h in handles {
            h.await.unwrap();
        }

        // Result should be valid (either None or a reasonable value)
        let latency = state.last_latency_ms();
        match latency {
            None => {} // Valid
            Some(v) => assert!(v <= 901, "Latency {} out of expected range", v),
        }
    }

    #[tokio::test]
    async fn test_disabled_at_concurrent_disable_enable() {
        let state = Arc::new(WorkerState::new(test_config("test")));
        let mut handles = vec![];

        for i in 0..10u32 {
            let s = Arc::clone(&state);
            handles.push(tokio::spawn(async move {
                if i % 2 == 0 {
                    s.disable(Some(format!("reason {i}"))).await;
                } else {
                    s.enable().await;
                }
            }));
        }

        for h in handles {
            h.await.unwrap();
        }

        // State should be consistent: if disabled, disabled_at is Some; if not, None
        let is_disabled = state.is_disabled().await;
        let disabled_at = state.disabled_at();
        if is_disabled {
            assert!(
                disabled_at.is_some(),
                "disabled worker should have disabled_at set"
            );
        }
        // Note: disabled_at might not be None when !is_disabled due to race
        // between status and atomic checks, which is acceptable for diagnostic data
    }

    #[tokio::test]
    async fn test_slot_reservation() {
        let config = WorkerConfig {
            id: WorkerId::new("test"),
            host: "localhost".to_string(),
            user: "user".to_string(),
            identity_file: "~/.ssh/id_rsa".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        };

        let state = WorkerState::new(config);
        assert_eq!(state.available_slots().await, 8);

        assert!(state.reserve_slots(4).await);
        assert_eq!(state.available_slots().await, 4);

        assert!(state.reserve_slots(4).await);
        assert_eq!(state.available_slots().await, 0);

        assert!(!state.reserve_slots(1).await); // Should fail

        state.release_slots(4).await;
        assert_eq!(state.available_slots().await, 4);
    }

    #[tokio::test]
    async fn test_disk_slots_derate_at_floor_and_recover() {
        let state = WorkerState::new(test_config("disk-slots"));
        assert_eq!(state.effective_total_slots().await, 8);
        for (free_gb, expected) in [
            (100.0, 8),
            (80.0, 7),
            (40.0, 3),
            (20.0, 1),
            (19.9, 0),
            (10.0, 0),
            (0.0, 0),
            (-1.0, 0),
            (f64::MAX, 8),
        ] {
            state
                .set_pressure_assessment(PressureAssessment {
                    disk_free_gb: Some(free_gb),
                    ..Default::default()
                })
                .await;
            assert_eq!(
                state.effective_total_slots().await,
                expected,
                "free GiB={free_gb}"
            );
            assert_eq!(state.available_slots().await, expected);
            assert_eq!(state.config.read().await.total_slots, 8);
        }
        for unknown in [None, Some(f64::NAN), Some(f64::INFINITY)] {
            state
                .set_pressure_assessment(PressureAssessment {
                    disk_free_gb: unknown,
                    ..Default::default()
                })
                .await;
            assert_eq!(state.effective_total_slots().await, 8);
        }
    }

    #[tokio::test]
    async fn test_disk_slots_keep_active_reservations_when_capacity_shrinks() {
        let state = WorkerState::new(test_config("disk-shrink"));
        assert!(state.reserve_slots(6).await);
        state
            .set_pressure_assessment(PressureAssessment {
                disk_free_gb: Some(20.0),
                ..Default::default()
            })
            .await;
        assert_eq!(state.effective_total_slots().await, 1);
        assert_eq!(state.used_slots(), 6);
        assert_eq!(state.available_slots().await, 0);
        assert!(!state.reserve_slots(1).await);
        state.release_slots(2).await;
        assert_eq!(state.used_slots(), 4);
        assert_eq!(state.available_slots().await, 0);
        state
            .set_pressure_assessment(PressureAssessment {
                disk_free_gb: Some(70.0),
                ..Default::default()
            })
            .await;
        assert_eq!(state.available_slots().await, 2);
        state.release_slots(4).await;
        assert_eq!(state.available_slots().await, 6);
    }

    #[tokio::test]
    async fn test_disk_slots_budget_survives_worker_reload_and_addition() {
        let config = rch_common::SelectionConfig {
            min_free_gb: Some(5.0),
            disk_gb_per_slot: 2.5,
            ..Default::default()
        };
        let pool = WorkerPool::with_selection_config(&config);
        pool.add_worker(test_config("first")).await;
        let first = pool.get(&WorkerId::new("first")).await.unwrap();
        first
            .set_pressure_assessment(PressureAssessment {
                disk_free_gb: Some(12.5),
                ..Default::default()
            })
            .await;
        assert_eq!(first.effective_total_slots().await, 3);
        let mut reloaded = test_config("first");
        reloaded.total_slots = 2;
        pool.add_worker(reloaded).await;
        assert_eq!(first.effective_total_slots().await, 2);
        pool.add_worker(test_config("later")).await;
        let later = pool.get(&WorkerId::new("later")).await.unwrap();
        later
            .set_pressure_assessment(first.pressure_assessment().await)
            .await;
        assert_eq!(later.effective_total_slots().await, 3);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_disk_slots_concurrent_reservations_respect_capacity() {
        let state = Arc::new(WorkerState::new(test_config("disk-race")));
        state
            .set_pressure_assessment(PressureAssessment {
                disk_free_gb: Some(40.0),
                ..Default::default()
            })
            .await;
        let barrier = Arc::new(tokio::sync::Barrier::new(16));
        let mut tasks = Vec::new();
        for _ in 0..16 {
            let worker = state.clone();
            let barrier = barrier.clone();
            tasks.push(tokio::spawn(async move {
                barrier.wait().await;
                worker.reserve_slots(1).await
            }));
        }
        let mut reserved = 0;
        for task in tasks {
            reserved += u32::from(task.await.unwrap());
        }
        assert_eq!(reserved, 3);
        assert_eq!(state.used_slots(), 3);
        assert_eq!(state.available_slots().await, 0);
        state.release_slots(reserved).await;
        assert_eq!(state.available_slots().await, 3);
    }

    #[tokio::test]
    async fn test_slot_reservation_rejects_integer_overflow() {
        let mut config = test_config("slot-overflow");
        config.total_slots = u32::MAX;
        let state = WorkerState::new(config);
        assert!(state.reserve_slots(u32::MAX).await);
        assert!(!state.reserve_slots(1).await);
        assert_eq!(state.used_slots(), u32::MAX);
    }

    #[tokio::test]
    async fn test_zero_slot_reservation_is_rejected() {
        let state = WorkerState::new(test_config("test"));

        assert!(!state.reserve_slots(0).await);
        assert_eq!(state.used_slots(), 0);
        assert_eq!(state.available_slots().await, 8);
    }

    #[tokio::test]
    async fn test_slot_reservation_exact_boundary() {
        let state = WorkerState::new(test_config("test"));

        // Reserve exactly all slots
        assert!(state.reserve_slots(8).await);
        assert_eq!(state.available_slots().await, 0);
        assert_eq!(state.used_slots(), 8);

        // Trying to reserve even 0 more fails when full
        assert!(!state.reserve_slots(1).await);

        // Release all
        state.release_slots(8).await;
        assert_eq!(state.available_slots().await, 8);
    }

    #[tokio::test]
    async fn test_release_slots_more_than_used() {
        let state = WorkerState::new(test_config("test"));

        // Reserve 2 slots
        assert!(state.reserve_slots(2).await);
        assert_eq!(state.used_slots(), 2);

        // Release more than reserved - should saturate to 0
        state.release_slots(10).await;
        assert_eq!(state.used_slots(), 0);
        assert_eq!(state.available_slots().await, 8);
    }

    #[tokio::test]
    async fn test_status_transitions() {
        let state = WorkerState::new(test_config("test"));

        assert_eq!(state.status().await, WorkerStatus::Healthy);

        state.set_status(WorkerStatus::Degraded).await;
        assert_eq!(state.status().await, WorkerStatus::Degraded);

        state.set_status(WorkerStatus::Unreachable).await;
        assert_eq!(state.status().await, WorkerStatus::Unreachable);

        state.set_status(WorkerStatus::Healthy).await;
        assert_eq!(state.status().await, WorkerStatus::Healthy);
    }

    #[tokio::test]
    async fn test_reserve_slots_refuses_draining_worker() {
        // Regression: the selector's `healthy_workers()` filter runs *before*
        // `reserve_slots`, so an operator invoking `drain()` between the two
        // must still prevent the reservation. Previously reserve_slots only
        // compared used + count against total_slots and silently accepted on
        // drained workers — a build could land on a worker we had just asked
        // to stop accepting work, breaking the drain contract.
        let state = WorkerState::new(test_config("drain-race"));
        assert!(state.reserve_slots(1).await, "fresh worker should accept");
        state.release_slots(1).await;

        state.drain().await;
        assert!(
            !state.reserve_slots(1).await,
            "draining worker must refuse new reservations"
        );
        assert_eq!(state.used_slots(), 0);

        state.check_drain_complete().await;
        assert_eq!(state.status().await, WorkerStatus::Drained);
        assert!(
            !state.reserve_slots(1).await,
            "drained worker must refuse new reservations"
        );

        state.disable(Some("maintenance".to_string())).await;
        assert!(
            !state.reserve_slots(1).await,
            "disabled worker must refuse new reservations"
        );
    }

    #[tokio::test]
    async fn test_apply_health_status_updates_non_administrative_states() {
        let state = WorkerState::new(test_config("test"));

        assert_eq!(
            state.apply_health_status(WorkerStatus::Degraded).await,
            WorkerStatus::Degraded
        );
        assert_eq!(state.status().await, WorkerStatus::Degraded);

        assert_eq!(
            state.apply_health_status(WorkerStatus::Unreachable).await,
            WorkerStatus::Unreachable
        );
        assert_eq!(state.status().await, WorkerStatus::Unreachable);

        assert_eq!(
            state.apply_health_status(WorkerStatus::Healthy).await,
            WorkerStatus::Healthy
        );
        assert_eq!(state.status().await, WorkerStatus::Healthy);
    }

    #[tokio::test]
    async fn test_apply_health_status_preserves_administrative_states() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.reserve_slots(1).await);
        state.drain().await;
        assert_eq!(
            state.apply_health_status(WorkerStatus::Healthy).await,
            WorkerStatus::Draining
        );
        assert_eq!(state.status().await, WorkerStatus::Draining);

        state.release_slots(1).await;
        assert_eq!(state.status().await, WorkerStatus::Drained);
        assert_eq!(
            state.apply_health_status(WorkerStatus::Healthy).await,
            WorkerStatus::Drained
        );
        assert_eq!(state.status().await, WorkerStatus::Drained);

        state.disable(Some("maintenance".to_string())).await;
        assert_eq!(
            state.apply_health_status(WorkerStatus::Healthy).await,
            WorkerStatus::Disabled
        );
        assert_eq!(state.status().await, WorkerStatus::Disabled);
    }

    #[tokio::test]
    async fn test_enter_bypass_excludes_from_scheduling() {
        // A worker quarantined into TemporaryBypass must look `Unreachable` to
        // every legacy consumer, report not-schedulable, and refuse new slot
        // reservations — all derived from the single lifecycle source of truth.
        let state = WorkerState::new(test_config("bypassed"));
        assert!(state.reserve_slots(1).await);
        state.release_slots(1).await;

        state.enter_bypass(BypassFailureClass::Ssh).await;

        assert_eq!(state.eligibility().await, EligibilityState::TemporaryBypass);
        assert_eq!(state.status().await, WorkerStatus::Unreachable);
        assert!(!state.lifecycle().await.is_schedulable());
        assert_eq!(
            state.lifecycle().await.bypass_cause,
            Some(BypassFailureClass::Ssh)
        );
        assert!(
            !state.reserve_slots(1).await,
            "a bypassed worker must refuse new reservations"
        );
    }

    #[tokio::test]
    async fn test_health_probe_never_clears_bypass() {
        // The cardinal safety invariant: lucky health observations routed
        // through the health monitor must NOT auto-rejoin a bypassed worker.
        // Only the recovery probe/canary loop may clear a quarantine.
        let state = WorkerState::new(test_config("flapper"));
        state
            .enter_bypass(BypassFailureClass::DiskInodePressure)
            .await;

        for _ in 0..5 {
            let effective = state.apply_health_status(WorkerStatus::Healthy).await;
            assert_eq!(effective, WorkerStatus::Unreachable);
            assert_eq!(state.eligibility().await, EligibilityState::TemporaryBypass);
        }
    }

    #[tokio::test]
    async fn test_recover_to_canary_then_promote() {
        // The full recovery path: bypass -> canary-pending -> healthy. The
        // canary-pending worker is still excluded from *normal* scheduling
        // (status reads Unreachable, reservations refused) while
        // is_canary_pending gates the single allowed canary build.
        let state = WorkerState::new(test_config("recoverer"));
        state.enter_bypass(BypassFailureClass::CircuitBreaker).await;

        state
            .recover_to_canary()
            .await
            .expect("bypass -> canary is legal");
        assert!(state.is_canary_pending().await);
        assert_eq!(
            state.eligibility().await,
            EligibilityState::RecoveredPendingCanary
        );
        assert_eq!(state.status().await, WorkerStatus::Unreachable);
        assert!(
            !state.reserve_slots(1).await,
            "canary-pending worker must not take normal builds"
        );

        state
            .promote_from_canary()
            .await
            .expect("canary-pending -> healthy is legal");
        assert_eq!(state.status().await, WorkerStatus::Healthy);
        assert!(state.lifecycle().await.is_schedulable());
        assert!(!state.is_canary_pending().await);
        assert!(state.reserve_slots(1).await, "rejoined worker accepts work");
    }

    #[tokio::test]
    async fn test_recover_to_canary_illegal_from_healthy() {
        // recover_to_canary is only legal from TemporaryBypass; a healthy worker
        // can never shortcut into the canary gate.
        let state = WorkerState::new(test_config("healthy"));
        let err = state.recover_to_canary().await.unwrap_err();
        assert_eq!(err.from, EligibilityState::Healthy);
        assert_eq!(err.to, EligibilityState::RecoveredPendingCanary);
    }

    #[tokio::test]
    async fn test_enable_clears_bypass() {
        // Operator re-enable is an authoritative override that clears a
        // quarantine (both axes back to Active + Healthy).
        let state = WorkerState::new(test_config("reenabled"));
        state
            .enter_bypass(BypassFailureClass::RuntimeToolchain)
            .await;
        assert_eq!(state.status().await, WorkerStatus::Unreachable);

        state.enable().await;
        assert_eq!(state.status().await, WorkerStatus::Healthy);
        assert!(state.lifecycle().await.is_schedulable());
        assert_eq!(state.lifecycle().await.bypass_cause, None);
    }

    #[tokio::test]
    async fn test_pool_healthy_workers_excludes_bypassed() {
        // The pool's healthy_workers() filter (the selector's candidate source)
        // must drop TemporaryBypass and RecoveredPendingCanary workers just like
        // an Unreachable one.
        let pool = WorkerPool::new();
        pool.add_worker(test_config("healthy")).await;
        pool.add_worker(test_config("bypassed")).await;
        pool.add_worker(test_config("canary")).await;

        pool.get(&WorkerId::new("bypassed"))
            .await
            .unwrap()
            .enter_bypass(BypassFailureClass::Ssh)
            .await;
        let canary = pool.get(&WorkerId::new("canary")).await.unwrap();
        canary.enter_bypass(BypassFailureClass::Ssh).await;
        canary.recover_to_canary().await.unwrap();

        let healthy = pool.healthy_workers().await;
        assert_eq!(healthy.len(), 1);
        assert_eq!(healthy[0].config.read().await.id, WorkerId::new("healthy"));
    }

    #[tokio::test]
    async fn test_disable_enable_cycle() {
        let state = WorkerState::new(test_config("test"));

        // Disable with reason
        state.disable(Some("Maintenance".to_string())).await;
        assert!(state.is_disabled().await);
        assert_eq!(state.status().await, WorkerStatus::Disabled);
        assert_eq!(
            state.disabled_reason().await,
            Some("Maintenance".to_string())
        );
        assert!(state.disabled_at().is_some());

        // Enable again
        state.enable().await;
        assert!(!state.is_disabled().await);
        assert_eq!(state.status().await, WorkerStatus::Healthy);
        assert!(state.disabled_reason().await.is_none());
        assert!(state.disabled_at().is_none());
    }

    #[tokio::test]
    async fn test_disable_without_reason() {
        let state = WorkerState::new(test_config("test"));

        state.disable(None).await;
        assert!(state.is_disabled().await);
        assert!(state.disabled_reason().await.is_none());
        assert!(state.disabled_at().is_some());
    }

    #[tokio::test]
    async fn test_drain_state() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.reserve_slots(1).await);
        state.drain().await;
        assert!(state.is_draining().await);
        assert_eq!(state.status().await, WorkerStatus::Draining);
        assert!(!state.is_disabled().await);

        // Enable clears draining
        state.enable().await;
        assert!(!state.is_draining().await);
        assert_eq!(state.status().await, WorkerStatus::Healthy);
    }

    #[tokio::test]
    async fn test_drained_state() {
        let state = WorkerState::new(test_config("test"));

        // No slots are in use, so drain completes immediately.
        state.drain().await;
        assert!(!state.is_draining().await);
        assert!(state.is_drained().await);
        assert_eq!(state.status().await, WorkerStatus::Drained);

        // Enable clears drained
        state.enable().await;
        assert!(!state.is_drained().await);
        assert_eq!(state.status().await, WorkerStatus::Healthy);
    }

    #[tokio::test]
    async fn test_drain_complete_with_active_jobs() {
        let state = WorkerState::new(test_config("test"));

        // Reserve a slot to simulate an active job
        assert!(state.reserve_slots(1).await);
        assert_eq!(state.used_slots(), 1);

        state.drain().await;
        assert!(state.is_draining().await);

        // With active jobs, calling check_drain_complete() should NOT transition to Drained
        state.check_drain_complete().await;
        assert!(state.is_draining().await);
        assert!(!state.is_drained().await);

        // Release the slot (job completes) - this automatically calls check_drain_complete()
        // and should transition to Drained since no jobs remain
        state.release_slots(1).await;
        assert_eq!(state.used_slots(), 0);

        // Worker should now be Drained (automatic transition via release_slots)
        assert!(!state.is_draining().await);
        assert!(state.is_drained().await);
        assert_eq!(state.status().await, WorkerStatus::Drained);
    }

    #[tokio::test]
    async fn test_drain_then_disable_disables_after_active_jobs_finish() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.reserve_slots(1).await);
        state
            .drain_then_disable(Some("maintenance".to_string()))
            .await;
        assert_eq!(state.status().await, WorkerStatus::Draining);
        assert!(state.disabled_reason().await.is_none());
        assert!(state.disabled_at().is_none());

        state.release_slots(1).await;

        assert_eq!(state.status().await, WorkerStatus::Disabled);
        assert_eq!(
            state.disabled_reason().await,
            Some("maintenance".to_string())
        );
        assert!(state.disabled_at().is_some());
        assert!(!state.remove_after_drain_pending().await);
    }

    #[tokio::test]
    async fn test_cached_projects() {
        let state = WorkerState::new(test_config("test"));

        assert!(!state.has_cached_project("project-a").await);

        state.add_cached_project("project-a".to_string()).await;
        assert!(state.has_cached_project("project-a").await);
        assert!(!state.has_cached_project("project-b").await);

        // Adding same project again should not duplicate
        state.add_cached_project("project-a".to_string()).await;
        let projects = state.cached_projects.read().await;
        assert_eq!(projects.len(), 1);
        drop(projects);

        state.add_cached_project("project-b".to_string()).await;
        assert!(state.has_cached_project("project-b").await);
        let projects = state.cached_projects.read().await;
        assert_eq!(projects.len(), 2);
    }

    #[tokio::test]
    async fn test_speed_score() {
        let state = WorkerState::new(test_config("test"));

        // Default is 50.0
        assert_eq!(state.get_speed_score(), 50.0);

        state.set_speed_score(85.5);
        assert_eq!(state.get_speed_score(), 85.5);

        state.set_speed_score(0.0);
        assert_eq!(state.get_speed_score(), 0.0);

        state.set_speed_score(100.0);
        assert_eq!(state.get_speed_score(), 100.0);
    }

    #[tokio::test]
    async fn test_latency_tracking() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.last_latency_ms().is_none());

        state.set_last_latency_ms(Some(42));
        assert_eq!(state.last_latency_ms(), Some(42));

        state.set_last_latency_ms(Some(100));
        assert_eq!(state.last_latency_ms(), Some(100));

        state.set_last_latency_ms(None);
        assert!(state.last_latency_ms().is_none());
    }

    #[tokio::test]
    async fn test_error_tracking() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.last_error().await.is_none());

        state.set_error("Connection refused".to_string()).await;
        assert_eq!(
            state.last_error().await,
            Some("Connection refused".to_string())
        );

        // Recording failure with error message updates last_error
        state.record_failure(Some("Timeout".to_string())).await;
        assert_eq!(state.last_error().await, Some("Timeout".to_string()));

        // Recording failure without message keeps previous error
        state.record_failure(None).await;
        assert_eq!(state.last_error().await, Some("Timeout".to_string()));
    }

    #[tokio::test]
    async fn test_circuit_breaker_basic() {
        let state = WorkerState::new(test_config("test"));

        // Initial state is Closed
        assert_eq!(state.circuit_state().await, Some(CircuitState::Closed));

        // Record success
        state.record_success().await;
        assert_eq!(state.circuit_state().await, Some(CircuitState::Closed));

        // Open circuit
        state.open_circuit().await;
        assert_eq!(state.circuit_state().await, Some(CircuitState::Open));

        // Half-open
        state.half_open_circuit().await;
        assert_eq!(state.circuit_state().await, Some(CircuitState::HalfOpen));

        // Close circuit clears error
        state.set_error("Test error".to_string()).await;
        state.close_circuit().await;
        assert_eq!(state.circuit_state().await, Some(CircuitState::Closed));
        assert!(state.last_error().await.is_none());
    }

    #[tokio::test]
    async fn test_capabilities() {
        let state = WorkerState::new(test_config("test"));

        // Default capabilities: nothing installed
        assert!(!state.has_bun().await);
        assert!(!state.has_node().await);
        assert!(!state.has_rust().await);
        assert!(!state.has_nix().await);

        // Set capabilities with Rust
        let mut caps = WorkerCapabilities::new();
        caps.rustc_version = Some("1.87.0-nightly".to_string());
        caps.disk_free_gb = Some(4.0);
        caps.disk_total_gb = Some(120.0);
        state.set_capabilities(caps).await;

        assert!(state.has_rust().await);
        assert!(!state.has_bun().await);
        assert!(!state.has_node().await);
        let pressure = state.pressure_assessment().await;
        assert_eq!(
            pressure.state,
            crate::disk_pressure::PressureState::Critical
        );
        assert_eq!(
            pressure.reason_code,
            "disk_critical_without_fresh_telemetry"
        );

        // Set capabilities with all runtimes
        let mut caps = WorkerCapabilities::new();
        caps.rustc_version = Some("1.87.0".to_string());
        caps.bun_version = Some("1.2.0".to_string());
        caps.node_version = Some("22.0.0".to_string());
        state.set_capabilities(caps.clone()).await;

        assert!(state.has_rust().await);
        assert!(state.has_bun().await);
        assert!(state.has_node().await);

        // Verify capabilities retrieval
        let retrieved = state.capabilities().await;
        assert_eq!(retrieved.rustc_version, Some("1.87.0".to_string()));
        assert_eq!(retrieved.bun_version, Some("1.2.0".to_string()));
    }

    #[tokio::test]
    async fn test_capability_refresh_preserves_fresh_pressure_telemetry() {
        let state = WorkerState::new(test_config("test"));
        state
            .set_pressure_assessment(crate::disk_pressure::PressureAssessment {
                state: crate::disk_pressure::PressureState::Healthy,
                confidence: crate::disk_pressure::PressureConfidence::High,
                reason_code: "all_pressure_rules_within_threshold".to_string(),
                policy_rule: "fresh_telemetry".to_string(),
                disk_free_gb: Some(50.0),
                disk_total_gb: Some(100.0),
                disk_free_ratio: Some(0.5),
                disk_io_util_pct: Some(0.0),
                memory_pressure: Some(10.0),
                telemetry_age_secs: Some(10),
                telemetry_fresh: true,
                evaluated_at_unix_ms: current_unix_ms(),
            })
            .await;

        let mut caps = WorkerCapabilities::new();
        caps.rustc_version = Some("1.87.0-nightly".to_string());
        caps.disk_free_gb = Some(90.0);
        caps.disk_total_gb = Some(120.0);
        state.set_capabilities(caps).await;

        let pressure = state.pressure_assessment().await;
        assert_eq!(pressure.state, crate::disk_pressure::PressureState::Healthy);
        assert_eq!(pressure.reason_code, "all_pressure_rules_within_threshold");
        assert_eq!(pressure.disk_free_gb, Some(90.0));
        assert_eq!(pressure.disk_total_gb, Some(120.0));
        assert_eq!(pressure.disk_free_ratio, Some(0.75));
        assert_eq!(pressure.disk_io_util_pct, Some(0.0));
        assert_eq!(pressure.memory_pressure, Some(10.0));
        assert!(pressure.telemetry_fresh);
        assert!(pressure.telemetry_age_secs.unwrap_or(u64::MAX) <= 11);
    }

    #[tokio::test]
    async fn test_capability_refresh_critical_disk_overrides_fresh_pressure() {
        let state = WorkerState::new(test_config("test"));
        state
            .set_pressure_assessment(crate::disk_pressure::PressureAssessment {
                state: crate::disk_pressure::PressureState::Healthy,
                confidence: crate::disk_pressure::PressureConfidence::High,
                reason_code: "all_pressure_rules_within_threshold".to_string(),
                policy_rule: "fresh_telemetry".to_string(),
                disk_free_gb: Some(50.0),
                disk_total_gb: Some(100.0),
                disk_free_ratio: Some(0.5),
                disk_io_util_pct: Some(0.0),
                memory_pressure: Some(10.0),
                telemetry_age_secs: Some(10),
                telemetry_fresh: true,
                evaluated_at_unix_ms: current_unix_ms(),
            })
            .await;

        let mut caps = WorkerCapabilities::new();
        caps.rustc_version = Some("1.87.0-nightly".to_string());
        caps.disk_free_gb = Some(2.0);
        caps.disk_total_gb = Some(120.0);
        state.set_capabilities(caps).await;

        let pressure = state.pressure_assessment().await;
        assert_eq!(
            pressure.state,
            crate::disk_pressure::PressureState::Critical
        );
        assert_eq!(
            pressure.reason_code,
            "disk_critical_without_fresh_telemetry"
        );
        assert_eq!(pressure.disk_free_gb, Some(2.0));
        assert!(!pressure.telemetry_fresh);
    }

    #[tokio::test]
    async fn test_toolchain_preflight_cache_records_fail_closed_reason() {
        let state = WorkerState::new(test_config("test"));

        state
            .record_toolchain_preflight(
                "nightly-2026-04-30".to_string(),
                false,
                Some("cargo binary not applicable".to_string()),
            )
            .await;

        let cached = state
            .toolchain_preflight_status("nightly-2026-04-30")
            .await
            .expect("toolchain preflight verdict should be cached");

        assert!(!cached.usable);
        assert_eq!(
            cached.reason.as_deref(),
            Some("cargo binary not applicable")
        );
        assert!(cached.is_fresh(Duration::from_secs(60)));
    }

    #[tokio::test]
    async fn test_update_config() {
        let state = WorkerState::new(test_config("test"));

        {
            let config = state.config.read().await;
            assert_eq!(config.total_slots, 8);
            assert_eq!(config.priority, 100);
        }

        // Update config
        let mut new_config = test_config("test");
        new_config.total_slots = 16;
        new_config.priority = 200;
        state.update_config(new_config).await;

        let config = state.config.read().await;
        assert_eq!(config.total_slots, 16);
        assert_eq!(config.priority, 200);
    }

    #[tokio::test]
    async fn test_update_config_cancels_pending_removal_drain() {
        let state = WorkerState::new(test_config("test"));

        assert!(state.reserve_slots(1).await);
        state.drain_for_removal().await;
        assert_eq!(state.status().await, WorkerStatus::Draining);
        assert!(state.remove_after_drain_pending().await);

        let mut new_config = test_config("test");
        new_config.priority = 250;
        state.update_config(new_config).await;

        assert_eq!(state.status().await, WorkerStatus::Healthy);
        assert!(!state.remove_after_drain_pending().await);
        state.release_slots(1).await;
        assert_eq!(state.status().await, WorkerStatus::Healthy);
    }

    // ============== WorkerPool Tests ==============

    #[tokio::test]
    async fn test_pool_new_empty() {
        let pool = WorkerPool::new();
        assert!(pool.is_empty());
        assert_eq!(pool.len(), 0);
    }

    #[tokio::test]
    async fn test_pool_add_and_get() {
        let pool = WorkerPool::new();

        pool.add_worker(test_config("worker-1")).await;
        assert_eq!(pool.len(), 1);
        assert!(!pool.is_empty());

        let worker = pool.get(&WorkerId::new("worker-1")).await;
        assert!(worker.is_some());
        let worker = worker.unwrap();
        assert_eq!(worker.config.read().await.total_slots, 8);

        // Non-existent worker
        assert!(pool.get(&WorkerId::new("worker-2")).await.is_none());
    }

    #[tokio::test]
    async fn test_pool_add_duplicate_updates() {
        let pool = WorkerPool::new();

        pool.add_worker(test_config("worker-1")).await;
        assert_eq!(pool.len(), 1);

        // Adding same id again updates config, doesn't add new worker
        let mut updated_config = test_config("worker-1");
        updated_config.total_slots = 16;
        pool.add_worker(updated_config).await;

        assert_eq!(pool.len(), 1);

        let worker = pool.get(&WorkerId::new("worker-1")).await.unwrap();
        assert_eq!(worker.config.read().await.total_slots, 16);
    }

    #[tokio::test]
    async fn test_pool_remove_worker() {
        let pool = WorkerPool::new();

        pool.add_worker(test_config("worker-1")).await;
        pool.add_worker(test_config("worker-2")).await;
        assert_eq!(pool.len(), 2);

        // Remove existing
        assert!(pool.remove_worker(&WorkerId::new("worker-1")).await);
        assert_eq!(pool.len(), 1);
        assert!(pool.get(&WorkerId::new("worker-1")).await.is_none());
        assert!(pool.get(&WorkerId::new("worker-2")).await.is_some());

        // Remove non-existent
        assert!(!pool.remove_worker(&WorkerId::new("worker-1")).await);
        assert_eq!(pool.len(), 1);
    }

    #[tokio::test]
    async fn test_pool_all_workers() {
        let pool = WorkerPool::new();

        pool.add_worker(test_config("worker-1")).await;
        pool.add_worker(test_config("worker-2")).await;

        // Disable one
        pool.get(&WorkerId::new("worker-1"))
            .await
            .unwrap()
            .disable(None)
            .await;

        let all = pool.all_workers().await;
        assert_eq!(all.len(), 2);
    }

    #[tokio::test]
    async fn test_pool_healthy_workers() {
        let pool = WorkerPool::new();

        pool.add_worker(test_config("healthy")).await;
        pool.add_worker(test_config("degraded")).await;
        pool.add_worker(test_config("unreachable")).await;
        pool.add_worker(test_config("disabled")).await;
        pool.add_worker(test_config("draining")).await;

        // Set statuses
        pool.get(&WorkerId::new("degraded"))
            .await
            .unwrap()
            .set_status(WorkerStatus::Degraded)
            .await;
        pool.get(&WorkerId::new("unreachable"))
            .await
            .unwrap()
            .set_status(WorkerStatus::Unreachable)
            .await;
        pool.get(&WorkerId::new("disabled"))
            .await
            .unwrap()
            .disable(None)
            .await;
        pool.get(&WorkerId::new("draining"))
            .await
            .unwrap()
            .drain()
            .await;

        let healthy = pool.healthy_workers().await;

        // Should include Healthy and Degraded only
        assert_eq!(healthy.len(), 2);

        // Collect worker ids manually
        let mut ids = Vec::new();
        for w in &healthy {
            ids.push(w.config.read().await.id.clone());
        }

        assert!(ids.contains(&WorkerId::new("healthy")));
        assert!(ids.contains(&WorkerId::new("degraded")));
    }

    #[tokio::test]
    async fn test_pool_set_status() {
        let pool = WorkerPool::new();
        pool.add_worker(test_config("worker-1")).await;

        pool.set_status(&WorkerId::new("worker-1"), WorkerStatus::Unreachable)
            .await;

        let worker = pool.get(&WorkerId::new("worker-1")).await.unwrap();
        assert_eq!(worker.status().await, WorkerStatus::Unreachable);

        // Setting status on non-existent worker is a no-op
        pool.set_status(&WorkerId::new("nonexistent"), WorkerStatus::Healthy)
            .await;
    }

    #[tokio::test]
    async fn test_pool_release_slots() {
        let pool = WorkerPool::new();
        pool.add_worker(test_config("worker-1")).await;

        let worker = pool.get(&WorkerId::new("worker-1")).await.unwrap();
        worker.reserve_slots(4).await;
        assert_eq!(worker.available_slots().await, 4);

        pool.release_slots(&WorkerId::new("worker-1"), 2).await;
        assert_eq!(worker.available_slots().await, 6);

        // Release on non-existent worker is a no-op
        pool.release_slots(&WorkerId::new("nonexistent"), 10).await;
    }

    #[tokio::test]
    async fn test_prune_drained() {
        let pool = WorkerPool::new();

        // Active worker
        let active = WorkerState::new(WorkerConfig {
            id: WorkerId::new("active"),
            host: "localhost".to_string(),
            user: "u".to_string(),
            identity_file: "i".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });
        pool.add_worker_state(active).await;

        // User-drained worker with 0 slots used. This must remain in the pool
        // so `rch workers enable <id>` can bring it back.
        let user_drained_empty = WorkerState::new(WorkerConfig {
            id: WorkerId::new("user_drained_empty"),
            host: "localhost".to_string(),
            user: "u".to_string(),
            identity_file: "i".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });
        user_drained_empty.drain().await;
        pool.add_worker_state(user_drained_empty).await;

        // Config-removal drain with 0 slots used. This is the only kind of
        // drained worker the background cleanup should prune.
        let removed_empty = WorkerState::new(WorkerConfig {
            id: WorkerId::new("removed_empty"),
            host: "localhost".to_string(),
            user: "u".to_string(),
            identity_file: "i".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });
        removed_empty.drain_for_removal().await;
        pool.add_worker_state(removed_empty).await;

        // Config-removal drain with slots in use (should NOT be pruned yet).
        //
        // In production, slot reservation happens while the worker is
        // still healthy (selection path) and a subsequent removal drain just
        // stops new reservations without releasing the in-flight ones.
        // `reserve_slots` (correctly) refuses after drain, so the test
        // must mirror that ordering.
        let removed_busy = WorkerState::new(WorkerConfig {
            id: WorkerId::new("removed_busy"),
            host: "localhost".to_string(),
            user: "u".to_string(),
            identity_file: "i".to_string(),
            total_slots: 8,
            priority: 100,
            tags: vec![],
            tools: Vec::new(),
        });
        assert!(removed_busy.reserve_slots(1).await);
        removed_busy.drain_for_removal().await;
        pool.add_worker_state(removed_busy).await;

        assert_eq!(pool.len(), 4);

        let pruned = pool.prune_drained().await;
        assert_eq!(pruned, 1);
        assert_eq!(pool.len(), 3);

        assert!(pool.get(&WorkerId::new("active")).await.is_some());
        assert!(
            pool.get(&WorkerId::new("user_drained_empty"))
                .await
                .is_some()
        );
        assert!(pool.get(&WorkerId::new("removed_busy")).await.is_some());
        assert!(pool.get(&WorkerId::new("removed_empty")).await.is_none());

        let removed_busy = pool.get(&WorkerId::new("removed_busy")).await.unwrap();
        removed_busy.release_slots(1).await;
        assert_eq!(pool.prune_drained().await, 1);
        assert!(pool.get(&WorkerId::new("removed_busy")).await.is_none());
    }

    #[tokio::test]
    async fn test_prune_drained_empty_pool() {
        let pool = WorkerPool::new();
        let pruned = pool.prune_drained().await;
        assert_eq!(pruned, 0);
    }

    #[tokio::test]
    async fn test_prune_drained_no_draining_workers() {
        let pool = WorkerPool::new();
        pool.add_worker(test_config("worker-1")).await;
        pool.add_worker(test_config("worker-2")).await;

        let pruned = pool.prune_drained().await;
        assert_eq!(pruned, 0);
        assert_eq!(pool.len(), 2);
    }

    #[tokio::test]
    async fn test_pool_default() {
        let pool = WorkerPool::default();
        assert!(pool.is_empty());
    }
}