eggress_runtime/

supervisor.rs

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use arc_swap::ArcSwap;
#[cfg(feature = "operations")]
use eggress_admin::{AdminSnapshot, AdminSnapshotProvider, ListenerInfo};
use eggress_core::listener::{is_listener_cancelled, TcpListener, TcpListenerConfig};
use eggress_core::ProtocolId;
use eggress_routing::health::HealthManager;
use eggress_routing::upstream::UpstreamRuntime;
use eggress_routing::{RouteService, SharedRoutingService};
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
use tracing::Instrument;

use crate::error::RuntimeError;
use crate::platform::{check_capability, PlatformCapability};
use crate::snapshot::{compile_runtime_snapshot, CompiledRuntimeSnapshot};

/// Pause between retries when a listener's `accept()` keeps failing (for
/// example fd exhaustion), so the loop does not tight-spin while the system
/// is already resource-starved.
const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100);

/// Log an accept failure, backing off briefly for persistent error classes so
/// repeated failures cannot hot-spin the accept loop. Transient races (a
/// queued connection vanishing before `accept`) retry immediately.
async fn handle_accept_error(context: &str, error: &std::io::Error) {
    match error.kind() {
        std::io::ErrorKind::WouldBlock
        | std::io::ErrorKind::Interrupted
        | std::io::ErrorKind::ConnectionAborted => {}
        _ => tokio::time::sleep(ACCEPT_ERROR_BACKOFF).await,
    }
    tracing::error!("{context} accept error: {error}");
}

/// Per-listener connection slot for listener implementations that cannot use
/// the core TCP listener wrapper (transparent and Unix sockets).
struct ListenerConnectionSlot {
    active: Arc<AtomicU64>,
}

impl ListenerConnectionSlot {
    fn try_acquire(active: &Arc<AtomicU64>, limit: u64) -> Option<Self> {
        active
            .fetch_update(Ordering::AcqRel, Ordering::Relaxed, |current| {
                (current < limit).then_some(current + 1)
            })
            .ok()
            .map(|_| Self {
                active: active.clone(),
            })
    }
}

impl Drop for ListenerConnectionSlot {
    fn drop(&mut self) {
        self.active.fetch_sub(1, Ordering::Release);
    }
}

/// Guard for active-connection accounting on non-TCP listeners.
///
/// TCP listeners account via `eggress_core::listener::PermitStream` (semaphore
/// permit held for the whole session). Transparent and Unix listeners use this
/// counter instead. The two mechanisms are **exclusive** — a connection must
/// use exactly one, never both, otherwise the `active_connections` metric
/// double-counts. Call sites below (transparent/Unix/QUIC paths) ensure this
/// invariant; TCP accept paths must not also create this guard.
///
/// Ordering: `AcqRel` on inc pairs with `Release` on dec for visibility of
/// the counter to the metrics reader. `Relaxed` would also be correct for a
/// metrics-only counter, but `AcqRel` is kept for consistency with the
/// previous implementation.
struct ActiveConnectionGuard {
    active: Arc<AtomicU64>,
}

impl ActiveConnectionGuard {
    fn new(active: Arc<AtomicU64>) -> Self {
        active.fetch_add(1, Ordering::AcqRel);
        Self { active }
    }
}

impl Drop for ActiveConnectionGuard {
    fn drop(&mut self) {
        self.active.fetch_sub(1, Ordering::Release);
    }
}

/// Result of a reload attempt.
#[derive(Debug)]
pub enum ReloadResult {
    /// Reload was applied successfully.
    Applied { generation: u64, upstreams: usize },
    /// Reload was rejected due to unsupported changes.
    Rejected { reason: String },
    /// Reload failed due to a compile or build error.
    Failed { error: String },
}

/// Adapter that exposes the runtime's compiled snapshot to the admin server.
///
/// Implements `AdminSnapshotProvider` so that admin handlers see live data:
/// each request reads the current `ArcSwap<CompiledRuntimeSnapshot>` rather
/// than a startup-captured copy. Reloads take effect on the next request.
#[cfg(feature = "operations")]
pub struct RuntimeAdminListenerInfos {
    state: Arc<RuntimeState>,
}

#[cfg(feature = "operations")]
struct RuntimeAdminState {
    snapshot: Arc<CompiledRuntimeSnapshot>,
    listener_addrs: Vec<Option<std::net::SocketAddr>>,
}

#[cfg(feature = "operations")]
impl AdminSnapshotProvider for RuntimeAdminListenerInfos {
    fn generation(&self) -> u64 {
        self.state.admin_snapshot.load().snapshot.generation
    }

    fn snapshot(&self) -> AdminSnapshot {
        let admin_state = self.state.admin_snapshot.load();
        let snap = &admin_state.snapshot;
        let addrs = &admin_state.listener_addrs;
        let listeners: Vec<ListenerInfo> = snap
            .listeners
            .iter()
            .enumerate()
            .map(|(idx, lcfg)| {
                let mode = if lcfg.transparent.as_ref().is_some_and(|t| t.enabled) {
                    Some("transparent".to_string())
                } else if lcfg.unix.is_some() {
                    Some("unix".to_string())
                } else {
                    Some("standard".to_string())
                };

                let (capability_status, original_dst_support) =
                    if lcfg.transparent.as_ref().is_some_and(|t| t.enabled) {
                        let cap = crate::platform::check_capability(
                            crate::platform::PlatformCapability::LinuxOriginalDstIpv4,
                        );
                        (
                            Some(cap.to_string()),
                            Some(cap == crate::platform::CapabilityStatus::Available),
                        )
                    } else {
                        (None, None)
                    };

                let (unix_socket_path, unix_socket_unlink_existing) =
                    if let Some(ref unix_cfg) = lcfg.unix {
                        (
                            Some(unix_cfg.path.display().to_string()),
                            Some(unix_cfg.unlink_existing),
                        )
                    } else {
                        (None, None)
                    };

                ListenerInfo {
                    name: lcfg.name.clone(),
                    bind: lcfg.bind.clone(),
                    local_addr: addrs
                        .get(idx)
                        .and_then(|a| *a)
                        .map(|a| a.to_string())
                        .or_else(|| unix_socket_path.clone())
                        .unwrap_or_default(),
                    protocols: lcfg.protocols.iter().map(|p| p.to_string()).collect(),
                    udp_enabled: lcfg.udp.as_ref().is_some_and(|u| u.enabled),
                    mode,
                    capability_status,
                    original_dst_support,
                    unix_socket_path,
                    unix_socket_unlink_existing,
                }
            })
            .collect();
        AdminSnapshot {
            generation: snap.generation,
            router: snap.router.clone(),
            pac: snap.admin.as_ref().and_then(|a| a.pac.clone()),
            static_routes: snap
                .admin
                .as_ref()
                .map(|a| a.static_content.clone())
                .unwrap_or_default(),
            listeners,
        }
    }
}

/// What is and isn't reloaded on SIGHUP:
///
/// **Reloaded (hot-swap, no downtime):**
/// - Upstream chains and health config (with Arc reuse for unchanged upstreams)
/// - Upstream groups, schedulers, and fallback policies
/// - Routing rules and default action
/// - Listener configuration metadata (name, bind, protocols, auth)
/// - Admin PAC and static content configuration
///
/// **NOT reloaded (requires full restart):**
/// - Listener socket bindings (bound before readiness)
/// - Process-level settings (log format, log level, shutdown grace)
/// - Timeout configuration
/// - Admin bind address
///
/// **UDP-specific reload semantics:**
/// - UDP limits apply to new associations only; existing associations keep their limits.
/// - UDP bind changes are restart-required.
/// - UDP advertise address changes are restart-required if socket bind changes.
/// - Route changes apply immediately to future UDP packets.
///
/// Classify whether a reload is supported given old and new listener
/// configs. Returns `Ok(())` if the reload is safe, or `Err(reason)`
/// if it should be rejected.
fn classify_listeners(
    old_listeners: &[eggress_config::compile::ListenerConfig],
    new_listeners: &[eggress_config::compile::ListenerConfig],
) -> Result<(), String> {
    if old_listeners.len() != new_listeners.len() {
        return Err(format!(
            "listener count changed ({} -> {}); restart required",
            old_listeners.len(),
            new_listeners.len()
        ));
    }

    for (old, new) in old_listeners.iter().zip(new_listeners.iter()) {
        if old.name != new.name {
            return Err(format!(
                "listener name changed ('{}' -> '{}'); restart required",
                old.name, new.name
            ));
        }
        if old.bind != new.bind {
            return Err(format!(
                "listener bind address changed for '{}': '{}' -> '{}'; restart required",
                old.name, old.bind, new.bind
            ));
        }
        match (&old.udp, &new.udp) {
            (Some(old_udp), Some(new_udp)) => {
                if old_udp.bind != new_udp.bind
                    || old_udp.enabled != new_udp.enabled
                    || old_udp.mode != new_udp.mode
                {
                    return Err(format!(
                        "UDP listener configuration changed for '{}'; restart required",
                        old.name
                    ));
                }
            }
            (None, Some(new_udp)) => {
                return Err(format!(
                    "UDP configuration added for '{}': '{}'; restart required",
                    new.name, new_udp.bind
                ));
            }
            (Some(_old_udp), None) => {
                return Err(format!(
                    "UDP configuration removed for '{}'; restart required",
                    old.name
                ));
            }
            (None, None) => {}
        }

        match (&old.transparent, &new.transparent) {
            (Some(old_t), Some(new_t)) => {
                if old_t.enabled != new_t.enabled {
                    return Err(format!(
                        "transparent config changed for '{}': enabled {} -> {}; restart required",
                        old.name, old_t.enabled, new_t.enabled
                    ));
                }
            }
            (None, Some(new_t)) => {
                if new_t.enabled {
                    return Err(format!(
                        "transparent proxy enabled for '{}'; restart required",
                        new.name
                    ));
                }
            }
            (Some(old_t), None) if old_t.enabled => {
                return Err(format!(
                    "transparent proxy configuration removed for '{}'; restart required",
                    old.name
                ));
            }
            (Some(_old_t), None) => {}
            (None, None) => {}
        }

        match (&old.unix, &new.unix) {
            (Some(old_u), Some(new_u)) => {
                if old_u.path != new_u.path {
                    return Err(format!(
                        "unix socket path changed for '{}': '{}' -> '{}'; restart required",
                        old.name,
                        old_u.path.display(),
                        new_u.path.display()
                    ));
                }
            }
            (None, Some(_new_u)) => {
                return Err(format!(
                    "unix socket added for '{}'; restart required",
                    new.name
                ));
            }
            (Some(_old_u), None) => {
                return Err(format!(
                    "unix socket removed for '{}'; restart required",
                    old.name
                ));
            }
            (None, None) => {}
        }

        // Hot-swappable listener metadata: log material changes so reload
        // observability is not silent. Socket topology is unchanged.
        if old.protocols != new.protocols {
            tracing::info!(
                listener = %old.name,
                old_protocols = ?old.protocols,
                new_protocols = ?new.protocols,
                "listener protocols changed via hot reload"
            );
        }
        // Auth material: compare presence and non-secret fields only.
        let old_auth_present = old.auth.is_some();
        let new_auth_present = new.auth.is_some();
        if old_auth_present != new_auth_present {
            tracing::info!(
                listener = %old.name,
                old_has_auth = old_auth_present,
                new_has_auth = new_auth_present,
                "listener auth presence changed via hot reload"
            );
        } else if let (Some(old_a), Some(new_a)) = (&old.auth, &new.auth) {
            if old_a.auth_type != new_a.auth_type || old_a.username != new_a.username {
                tracing::info!(
                    listener = %old.name,
                    "listener auth material changed via hot reload"
                );
            }
        }
        if old.tls.is_some() != new.tls.is_some() {
            tracing::info!(
                listener = %old.name,
                old_has_tls = old.tls.is_some(),
                new_has_tls = new.tls.is_some(),
                "listener TLS presence changed via hot reload"
            );
        }
        if old.shadowsocks.is_some() != new.shadowsocks.is_some()
            || old.trojan.is_some() != new.trojan.is_some()
        {
            tracing::info!(
                listener = %old.name,
                "listener shadowsocks/trojan presence changed via hot reload"
            );
        }
        if old.connection_limit != new.connection_limit
            || old.fixed_target != new.fixed_target
            || old.local_bind != new.local_bind
        {
            tracing::info!(
                listener = %old.name,
                "listener connection_limit/fixed_target/local_bind changed via hot reload"
            );
        }
    }

    Ok(())
}

pub fn classify_reload_config(
    old_listeners: &[eggress_config::compile::ListenerConfig],
    old_timeouts: &eggress_config::compile::TimeoutConfig,
    old_admin: Option<&eggress_config::compile::AdminConfig>,
    new_config: &eggress_config::compile::RuntimeConfig,
) -> Result<(), String> {
    classify_listeners(old_listeners, &new_config.listeners)?;
    if old_timeouts != &new_config.timeouts {
        return Err("timeout configuration changed; restart required".to_string());
    }

    let old_admin_endpoint = old_admin.map(|admin| (admin.enabled, admin.bind.as_str()));
    let new_admin_endpoint = new_config
        .admin
        .as_ref()
        .map(|admin| (admin.enabled, admin.bind.as_str()));
    if old_admin_endpoint != new_admin_endpoint {
        return Err("admin endpoint bind configuration changed; restart required".to_string());
    }

    Ok(())
}

struct PreparedListener {
    name: String,
    #[allow(dead_code)] // used only with `operations` feature
    bind: String,
    protocols: Vec<ProtocolId>,
    listener: TcpListener,
    local_addr: std::net::SocketAddr,
    auth: eggress_server::accept::InboundAuthentication,
    handshake_timeout: Duration,
    udp: Option<eggress_config::compile::CompiledListenerUdpConfig>,
    tls: Option<eggress_config::compile::CompiledListenerTlsConfig>,
    shadowsocks: Option<eggress_config::model::ShadowsocksListenerConfig>,
    trojan: Option<eggress_config::model::ListenerTrojanConfig>,
    fixed_target: Option<eggress_core::TargetAddr>,
    local_bind: Option<String>,
}

#[cfg(feature = "quic")]
struct PreparedQuicListener {
    name: String,
    protocols: Vec<ProtocolId>,
    listener: Arc<eggress_transport_quic::QuicListener>,
    local_addr: std::net::SocketAddr,
    auth: eggress_server::accept::InboundAuthentication,
    handshake_timeout: Duration,
    connection_limit: u64,
}

#[cfg(feature = "extended")]
type PreparedShadowsocksUdpRelay = (
    Arc<tokio::net::UdpSocket>,
    eggress_udp::standalone_shadowsocks::ShadowsocksStandaloneUdpConfig,
);

#[cfg(feature = "extended")]
async fn prepare_shadowsocks_udp_relay(
    prepared_listener: &PreparedListener,
    udp_cfg: &eggress_config::compile::CompiledListenerUdpConfig,
    routing: Arc<dyn RouteService>,
    state: &RuntimeState,
) -> Result<PreparedShadowsocksUdpRelay, RuntimeError> {
    let ss = prepared_listener.shadowsocks.as_ref().ok_or_else(|| {
        RuntimeError::Other(format!(
            "listener '{}' shadowsocks_udp mode requires shadowsocks config",
            prepared_listener.name
        ))
    })?;
    let method =
        eggress_protocol_shadowsocks::CipherMethod::parse_method(&ss.method).map_err(|e| {
            RuntimeError::Other(format!(
                "listener '{}' has invalid shadowsocks method '{}': {}",
                prepared_listener.name, ss.method, e
            ))
        })?;
    let socket = Arc::new(
        tokio::net::UdpSocket::bind(udp_cfg.bind)
            .await
            .map_err(|e| RuntimeError::ListenerBind {
                addr: udp_cfg.bind.to_string(),
                source: e,
            })?,
    );
    let local_addr = socket
        .local_addr()
        .map_err(|e| RuntimeError::ListenerBind {
            addr: udp_cfg.bind.to_string(),
            source: e,
        })?;
    tracing::info!(
        "shadowsocks UDP relay listening on {local_addr} ({})",
        prepared_listener.name
    );

    let relay_config = eggress_udp::standalone_shadowsocks::ShadowsocksStandaloneUdpConfig {
        routing,
        udp_metrics: state.udp_metrics.clone(),
        shadowsocks_metrics: Some(state.shadowsocks_metrics.clone()),
        limits: eggress_udp::limits::UdpLimits::from_listener_config(
            udp_cfg.max_associations_global,
            udp_cfg.max_associations,
            udp_cfg.max_targets_per_association,
            udp_cfg.max_datagram_size,
            udp_cfg.idle_timeout,
            udp_cfg.client_pin,
            udp_cfg.target_idle_timeout,
        ),
        listener: prepared_listener.name.clone(),
        generation: state.snapshot.load().generation,
        method,
        password: ss.password.clone(),
        allow_private_egress: udp_cfg.allow_private_egress,
    };

    Ok((socket, relay_config))
}

/// Compute the advertised IP for the SOCKS5 UDP ASSOCIATE reply.
///
/// Derivation rules:
/// 1. If `advertise` is configured, use it.
/// 2. Else if UDP bind IP is not unspecified, use UDP bind IP.
/// 3. Else if TCP peer is loopback, use loopback matching address family.
/// 4. Else return a config error requiring explicit `advertise`.
fn compute_advertise_ip(
    configured_advertise: Option<std::net::IpAddr>,
    udp_bind_ip: std::net::IpAddr,
    tcp_peer: Option<std::net::SocketAddr>,
) -> Result<std::net::IpAddr, eggress_udp::error::UdpError> {
    if let Some(ip) = configured_advertise {
        return Ok(ip);
    }

    if !udp_bind_ip.is_unspecified() {
        return Ok(udp_bind_ip);
    }

    if let Some(tcp_peer) = tcp_peer {
        let peer_is_loopback = tcp_peer.ip().is_loopback()
            || matches!(
                tcp_peer.ip(),
                std::net::IpAddr::V6(ipv6)
                    if ipv6
                        .to_ipv4_mapped()
                        .is_some_and(|ipv4| ipv4.is_loopback())
            );
        if peer_is_loopback {
            // Preserve the family selected by the unspecified UDP bind. This
            // matters for dual-stack sockets and IPv4-mapped IPv6 peers.
            match udp_bind_ip {
                std::net::IpAddr::V4(_) => {
                    return Ok(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST));
                }
                std::net::IpAddr::V6(_) => {
                    return Ok(std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST));
                }
            }
        }
    }

    Err(eggress_udp::error::UdpError::Other(
        "UDP relay requires explicit advertise address when bind is unspecified and client is not loopback".to_string()
    ))
}

struct RuntimeUdpService {
    _listener_name: String,
    udp_config: eggress_config::compile::CompiledListenerUdpConfig,
    registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
    metrics: Arc<dyn eggress_server::SessionMetrics>,
    udp_metrics: Arc<eggress_udp::metrics::UdpMetrics>,
    routing: Arc<SharedRoutingService>,
    udp_tasks: TaskTracker,
}

impl eggress_server::UdpService for RuntimeUdpService {
    fn create_association(
        &self,
        listener: &str,
        client_tcp_peer: Option<std::net::SocketAddr>,
        identity: eggress_core::ClientIdentity,
        generation: u64,
    ) -> std::pin::Pin<
        Box<
            dyn std::future::Future<
                    Output = Result<
                        eggress_server::UdpAssociationHandle,
                        eggress_udp::error::UdpError,
                    >,
                > + Send
                + 'static,
        >,
    > {
        let registry = self.registry.clone();
        let metrics = self.metrics.clone();
        let udp_metrics = self.udp_metrics.clone();
        let routing = self.routing.clone();
        let udp_tasks = self.udp_tasks.clone();
        let udp_config = self.udp_config.clone();
        let listener = listener.to_string();
        Box::pin(async move {
            let assoc = registry
                .create_association(&listener, client_tcp_peer, identity, generation)
                .await?;
            metrics.record_udp_association_created();

            let relay_socket =
                std::sync::Arc::new(tokio::net::UdpSocket::bind(udp_config.bind).await?);
            let local_addr = relay_socket.local_addr()?;

            let advertised_ip =
                compute_advertise_ip(udp_config.advertise, local_addr.ip(), client_tcp_peer)?;
            let relay_addr = std::net::SocketAddr::new(advertised_ip, local_addr.port());

            let relay_config = eggress_udp::relay::RelayConfig {
                routing: routing as Arc<dyn RouteService>,
                udp_metrics: udp_metrics.clone(),
                limits: eggress_udp::limits::UdpLimits::from_listener_config(
                    udp_config.max_associations_global,
                    udp_config.max_associations,
                    udp_config.max_targets_per_association,
                    udp_config.max_datagram_size,
                    udp_config.idle_timeout,
                    udp_config.client_pin,
                    udp_config.target_idle_timeout,
                ),
                listener: listener.clone(),
                generation,
                identity: assoc.meta.identity.clone(),
                client_tcp_peer,
                registry: registry.clone(),
                allow_private_egress: udp_config.allow_private_egress,
                upstream_connect_timeout: udp_config.upstream_connect_timeout,
                upstream_udp_bind: udp_config.upstream_udp_bind,
            };

            let relay_assoc = assoc.clone();
            let relay_cancel = assoc.cancel.clone();
            let assoc_id = assoc.id;
            let relay_udp_metrics = udp_metrics.clone();
            udp_tasks.spawn(async move {
                let result = eggress_udp::relay::udp_relay_loop(
                    relay_socket,
                    relay_assoc,
                    relay_config,
                    relay_cancel,
                )
                .await;
                if let Err(error) = result {
                    relay_udp_metrics.record_association_failure();
                    tracing::warn!(
                        %error,
                        association_id = ?assoc_id,
                        "UDP relay ended with error"
                    );
                }
            });

            Ok(eggress_server::UdpAssociationHandle {
                id: assoc.id,
                relay_addr,
                cancel: assoc.cancel.clone(),
            })
        })
    }

    fn is_enabled(&self) -> bool {
        self.udp_config.enabled
    }

    fn active_count(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = usize> + Send + 'static>> {
        let registry = self.registry.clone();
        Box::pin(async move { registry.active_count().await })
    }
}

pub struct RuntimeState {
    pub snapshot: Arc<ArcSwap<CompiledRuntimeSnapshot>>,
    pub routing: Arc<SharedRoutingService>,
    pub metrics: Arc<dyn eggress_server::SessionMetrics>,
    pub readiness: Arc<AtomicBool>,
    pub start_time: Instant,
    pub active_connections: Arc<AtomicU64>,
    pub connection_counter: Arc<AtomicU64>,
    pub admin_local_addr: Arc<Mutex<Option<std::net::SocketAddr>>>,
    pub listener_addrs: Arc<Mutex<Vec<Option<std::net::SocketAddr>>>>,
    #[cfg(feature = "operations")]
    admin_snapshot: Arc<ArcSwap<RuntimeAdminState>>,
    pub health: Arc<Mutex<Option<HealthManager>>>,
    pub health_cancel: CancellationToken,
    pub health_runtime: Mutex<Option<tokio::runtime::Handle>>,
    pub udp_registry: Arc<eggress_udp::registry::UdpAssociationRegistry>,
    pub udp_metrics: Arc<eggress_udp::metrics::UdpMetrics>,
    #[cfg(feature = "extended")]
    pub shadowsocks_metrics: Arc<eggress_protocol_shadowsocks::ShadowsocksMetrics>,
    pub udp_tasks: TaskTracker,
    pub transparent_accepted_total: Arc<AtomicU64>,
    pub transparent_original_dst_failed_total: Arc<AtomicU64>,
    #[cfg(feature = "reverse")]
    pub reverse_registry: Arc<eggress_admin::ReverseRegistry>,
    #[cfg(feature = "reverse")]
    pub reverse_metrics: Arc<eggress_protocol_reverse::metrics::ReverseMetrics>,
}

impl RuntimeState {
    pub fn generation(&self) -> u64 {
        self.snapshot.load().generation
    }

    #[cfg(feature = "operations")]
    fn publish_admin_snapshot(&self, snapshot: Arc<CompiledRuntimeSnapshot>) {
        let listener_addrs = self.admin_snapshot.load().listener_addrs.clone();
        self.admin_snapshot.store(Arc::new(RuntimeAdminState {
            snapshot,
            listener_addrs,
        }));
    }

    #[cfg(feature = "operations")]
    fn publish_admin_listener_addrs(
        &self,
        snapshot: Arc<CompiledRuntimeSnapshot>,
        listener_addrs: Vec<Option<std::net::SocketAddr>>,
    ) {
        self.admin_snapshot.store(Arc::new(RuntimeAdminState {
            snapshot,
            listener_addrs,
        }));
    }

    /// Restart health probes for the upstreams in the current snapshot.
    pub fn restart_health_probes(&self) {
        let mut guard = self.health.lock().unwrap_or_else(|error| {
            tracing::warn!("health manager state was poisoned; resetting it: {error}");
            let mut guard = error.into_inner();
            *guard = None;
            self.health.clear_poison();
            guard
        });
        if let Some(ref mut health) = *guard {
            health.stop_all();
        }
        let upstreams: Vec<Arc<UpstreamRuntime>> =
            self.snapshot.load().upstreams.values().cloned().collect();
        if !upstreams.is_empty() {
            let mut health = HealthManager::new(self.health_cancel.clone());
            if let Some(handle) = self
                .health_runtime
                .lock()
                .unwrap_or_else(|error| error.into_inner())
                .clone()
            {
                health.start_probes_on(&handle, &upstreams);
            }
            *guard = Some(health);
        } else {
            *guard = None;
        }
    }
}

#[allow(dead_code)]
pub struct ServiceSupervisor {
    config_path: Option<String>,
    state: Arc<RuntimeState>,
    metrics_registry: Arc<eggress_metrics::MetricsRegistry>,
    cancel: CancellationToken,
    listener_cancel: CancellationToken,
    connection_cancel: CancellationToken,
    health_cancel: CancellationToken,
    admin_cancel: CancellationToken,
    health: Arc<Mutex<Option<HealthManager>>>,
    tasks: TaskTracker,
    connection_tasks: TaskTracker,
    admin_tasks: TaskTracker,
    shutdown_grace: Duration,
    rt_config: eggress_config::compile::RuntimeConfig,
    tls_client_config: Option<std::sync::Arc<rustls::ClientConfig>>,
    #[cfg(feature = "ssh")]
    ssh_sessions: Arc<eggress_transport_ssh::SshSessionCache>,
    compatibility_options: CompatibilityOptions,
}

/// Explicit options used only by the pproxy compatibility executable.
/// Native configuration startup always uses the default value.
#[derive(Debug, Clone, Default)]
pub struct CompatibilityOptions {
    /// Whether the caller is the explicitly opt-in pproxy compatibility executable.
    pub compatibility_mode: bool,
    pub auth_timeout: Option<Duration>,
    pub system_proxy: bool,
    /// Compatibility-only debug propagation. Native Eggress never enables it.
    pub debug: bool,
    /// Compatibility verbosity count, used for human-readable session and
    /// traffic events backed by the normal session reports.
    pub verbose_level: u8,
}

impl ServiceSupervisor {
    pub fn start(config_path: &str) -> Result<Self, RuntimeError> {
        let (rt_config, warnings) = eggress_config::load_and_validate_with_warnings(config_path)
            .map_err(|e| RuntimeError::Config(e.to_string()))?;

        for warning in &warnings {
            tracing::warn!("config security warning: {warning}");
        }

        Self::init_with_config(
            rt_config,
            Some(config_path.to_string()),
            CompatibilityOptions::default(),
        )
    }

    /// Start from an already-validated [`RuntimeConfig`] without reading a file.
    ///
    /// When `config_path` is provided (file-backed startup), SIGHUP reload
    /// is enabled. When `None` (in-memory/compatibility startup), SIGHUP
    /// reload is disabled because there is no stable user-authored config
    /// file to reload from.
    pub fn start_from_config(
        rt_config: eggress_config::compile::RuntimeConfig,
        config_path: Option<String>,
    ) -> Result<Self, RuntimeError> {
        Self::init_with_config(rt_config, config_path, CompatibilityOptions::default())
    }

    pub fn start_from_config_with_options(
        rt_config: eggress_config::compile::RuntimeConfig,
        config_path: Option<String>,
        compatibility_options: CompatibilityOptions,
    ) -> Result<Self, RuntimeError> {
        Self::init_with_config(rt_config, config_path, compatibility_options)
    }

    fn init_with_config(
        rt_config: eggress_config::compile::RuntimeConfig,
        config_path: Option<String>,
        compatibility_options: CompatibilityOptions,
    ) -> Result<Self, RuntimeError> {
        #[cfg(not(feature = "reverse"))]
        if !rt_config.reverse_servers.is_empty() || !rt_config.reverse_clients.is_empty() {
            return Err(RuntimeError::Other(
                "reverse proxy support not included in this build".to_string(),
            ));
        }

        #[cfg(not(feature = "operations"))]
        if rt_config.admin.as_ref().is_some_and(|a| a.enabled) {
            return Err(RuntimeError::Other(
                "admin server support not included in this build; \
                 enable the 'operations' feature or remove [admin] from config"
                    .to_string(),
            ));
        }

        for lcfg in &rt_config.listeners {
            if lcfg.unix.is_none() {
                let _bind_addr: std::net::SocketAddr =
                    lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
                        addr: lcfg.bind.clone(),
                        source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
                    })?;
            }
        }

        let udp_metrics = Arc::new(eggress_udp::metrics::UdpMetrics::new());
        #[cfg(feature = "extended")]
        let shadowsocks_metrics = Arc::new(eggress_protocol_shadowsocks::ShadowsocksMetrics::new());

        let metrics_registry = Arc::new(eggress_metrics::MetricsRegistry::new());
        metrics_registry.set_udp_metrics(udp_metrics.clone());
        #[cfg(feature = "extended")]
        {
            metrics_registry.set_shadowsocks_metrics(shadowsocks_metrics.clone());
        }
        let metrics: Arc<dyn eggress_server::SessionMetrics> = metrics_registry.clone();
        let readiness = Arc::new(AtomicBool::new(false));

        let snapshot = compile_runtime_snapshot(&rt_config, None)
            .map_err(|e| RuntimeError::Config(e.to_string()))?;
        let snapshot = Arc::new(ArcSwap::from_pointee(snapshot));

        let routing = Arc::new(SharedRoutingService::new_arc(
            snapshot.load().router.clone(),
        ));

        let active_connections = Arc::new(AtomicU64::new(0));
        let connection_counter = Arc::new(AtomicU64::new(1));

        let mut udp_global_limit: Option<usize> = None;
        for listener in &rt_config.listeners {
            if let Some(udp) = &listener.udp {
                let value = udp.max_associations_global;
                match udp_global_limit {
                    None => udp_global_limit = Some(value),
                    Some(existing) if existing != value => {
                        tracing::warn!(
                            listener = %listener.name,
                            existing,
                            other = value,
                            "multiple listeners specify udp.max_associations_global with different values; using first"
                        );
                    }
                    Some(_) => {}
                }
            }
        }
        let udp_global_limit = udp_global_limit.unwrap_or(1024);

        let udp_registry = Arc::new(eggress_udp::registry::UdpAssociationRegistry::new(
            eggress_udp::limits::UdpLimits {
                max_associations_global: udp_global_limit,
                ..Default::default()
            },
        ));

        let cancel = CancellationToken::new();
        let listener_cancel = CancellationToken::new();
        let connection_cancel = CancellationToken::new();
        let health_cancel = CancellationToken::new();
        let admin_cancel = CancellationToken::new();
        let health = Arc::new(Mutex::new(if snapshot.load().upstreams.is_empty() {
            None
        } else {
            Some(HealthManager::new(health_cancel.clone()))
        }));

        #[cfg(feature = "reverse")]
        let reverse_metrics = Arc::new(eggress_protocol_reverse::metrics::ReverseMetrics::new());
        let udp_tasks = TaskTracker::new();

        let state = Arc::new(RuntimeState {
            snapshot: snapshot.clone(),
            routing: routing.clone(),
            metrics: metrics.clone(),
            readiness,
            start_time: Instant::now(),
            active_connections,
            connection_counter,
            admin_local_addr: Arc::new(Mutex::new(None)),
            listener_addrs: Arc::new(Mutex::new(Vec::new())),
            #[cfg(feature = "operations")]
            admin_snapshot: Arc::new(ArcSwap::from_pointee(RuntimeAdminState {
                snapshot: snapshot.load_full(),
                listener_addrs: Vec::new(),
            })),
            health: health.clone(),
            health_cancel: health_cancel.clone(),
            health_runtime: Mutex::new(None),
            udp_registry,
            udp_metrics,
            #[cfg(feature = "extended")]
            shadowsocks_metrics,
            udp_tasks: udp_tasks.clone(),
            transparent_accepted_total: Arc::new(AtomicU64::new(0)),
            transparent_original_dst_failed_total: Arc::new(AtomicU64::new(0)),
            #[cfg(feature = "reverse")]
            reverse_registry: Arc::new(eggress_admin::ReverseRegistry::new()),
            #[cfg(feature = "reverse")]
            reverse_metrics,
        });

        // Bridge transparent proxy atomics to MetricsRegistry for /metrics
        metrics_registry.set_transparent_counters(
            state.transparent_accepted_total.clone(),
            state.transparent_original_dst_failed_total.clone(),
        );

        #[cfg(feature = "ssh")]
        let ssh_sessions = Arc::new(if compatibility_options.compatibility_mode {
            let insecure_acknowledged = std::env::var("EGRESS_SSH_INSECURE_HOST_KEYS")
                .map(|value| matches!(value.as_str(), "1" | "true" | "yes"))
                .unwrap_or(false);
            if insecure_acknowledged {
                eggress_transport_ssh::SshSessionCache::new_compatibility()
            } else {
                tracing::warn!(
                    "compatibility mode would disable SSH host-key verification; \
                     keeping known_hosts verification enabled. To explicitly \
                     accept unverified SSH host keys (MITM risk), set \
                     EGRESS_SSH_INSECURE_HOST_KEYS=1"
                );
                eggress_transport_ssh::SshSessionCache::new()
            }
        } else {
            eggress_transport_ssh::SshSessionCache::new()
        });

        let tasks = TaskTracker::new();
        let connection_tasks = TaskTracker::new();

        let shutdown_grace = rt_config.process.shutdown_grace;

        Ok(ServiceSupervisor {
            config_path,
            state,
            metrics_registry,
            cancel,
            listener_cancel,
            connection_cancel,
            health_cancel,
            admin_cancel,
            health: health.clone(),
            tasks,
            connection_tasks,
            admin_tasks: TaskTracker::new(),
            shutdown_grace,
            rt_config,
            tls_client_config: None,
            #[cfg(feature = "ssh")]
            ssh_sessions,
            compatibility_options,
        })
    }

    pub fn state(&self) -> &Arc<RuntimeState> {
        &self.state
    }

    pub fn shutdown_token(&self) -> CancellationToken {
        self.cancel.clone()
    }

    /// Override the TLS client config used for upstream connections (e.g., Trojan).
    /// Intended for test-only use (e.g., insecure TLS for self-signed certs).
    #[allow(dead_code)]
    pub fn with_tls_client_config(mut self, config: std::sync::Arc<rustls::ClientConfig>) -> Self {
        self.tls_client_config = Some(config);
        self
    }

    /// Classify whether a reload is supported given old and new listener configs.
    /// Returns `Ok(())` if the reload is safe, or `Err(reason)` if it should be rejected.
    ///
    /// UDP-specific reload semantics:
    /// - UDP bind changes are restart-required.
    /// - UDP advertise address changes are restart-required if socket bind changes.
    /// - UDP limits apply to new associations only; existing ones keep their limits.
    /// - Route changes apply immediately to future UDP packets.
    fn classify_reload(
        &self,
        new_config: &eggress_config::compile::RuntimeConfig,
    ) -> Result<(), String> {
        classify_reload_config(
            &self.rt_config.listeners,
            &self.rt_config.timeouts,
            self.rt_config.admin.as_ref(),
            new_config,
        )
    }

    /// Attempt to reload configuration. Encapsulates the full reload transaction:
    /// 1. Load and compile new config
    /// 2. Classify unsupported changes (reject if listener topology changed)
    /// 3. Build new snapshot with previous snapshot for Arc reuse
    /// 4. Update stored rt_config for subsequent reloads
    /// 5. Atomically swap routing and snapshot
    pub fn reload_config(&mut self) -> ReloadResult {
        let config_path = match self.config_path {
            Some(ref p) => p.clone(),
            None => {
                return ReloadResult::Rejected {
                    reason: "no config file path available for reload".to_string(),
                };
            }
        };
        let new_rt_config = match eggress_config::compile::load_and_compile(&config_path) {
            Ok(c) => c,
            Err(e) => {
                return ReloadResult::Failed {
                    error: format!("config load: {e}"),
                };
            }
        };

        if let Err(reason) = self.classify_reload(&new_rt_config) {
            return ReloadResult::Rejected { reason };
        }

        let prev_snapshot = self.state.snapshot.load();
        let prev_ref: Option<&CompiledRuntimeSnapshot> = Some(&prev_snapshot);
        let new_snapshot = match compile_runtime_snapshot(&new_rt_config, prev_ref) {
            Ok(s) => s,
            Err(e) => {
                return ReloadResult::Failed {
                    error: format!("snapshot build: {e}"),
                };
            }
        };

        let upstream_count = new_snapshot.upstreams.len();
        let gen = new_snapshot.generation;

        // Keep the configuration used by the next reload in sync before making
        // the new snapshot visible to readers.
        self.rt_config = new_rt_config;

        // Snapshot must be published before the router swap. Readers that observe
        // the new generation via `snapshot.load()` pull the router from that
        // same snapshot Arc, so any reader seeing the new generation also sees
        // the router that belongs to it.
        let new_snapshot = Arc::new(new_snapshot);
        self.state.snapshot.store(new_snapshot.clone());
        self.state.routing.swap_arc(new_snapshot.router.clone());
        #[cfg(feature = "operations")]
        self.state.publish_admin_snapshot(new_snapshot.clone());

        self.state.restart_health_probes();
        eggress_protocol_http::H2_POOL_REGISTRY.clear();

        ReloadResult::Applied {
            generation: gen,
            upstreams: upstream_count,
        }
    }

    pub fn run(&mut self) -> Result<(), RuntimeError> {
        #[allow(unused_variables)]
        let config_path = self.config_path.clone().unwrap_or_default();
        let routing = self.state.routing.clone();
        let listener_cancel = self.listener_cancel.clone();
        let connection_cancel = self.connection_cancel.clone();
        let health_cancel = self.health_cancel.clone();
        let admin_cancel = self.admin_cancel.clone();
        let cancel = self.cancel.clone();
        #[allow(unused_variables)]
        let metrics = self.state.metrics.clone();
        let readiness = self.state.readiness.clone();
        #[cfg(feature = "operations")]
        let admin_state_ref = self.state.clone();
        let active_connections = self.state.active_connections.clone();
        let shutdown_grace = self.shutdown_grace;
        let tasks = self.tasks.clone();
        let connection_tasks = self.connection_tasks.clone();
        let admin_tasks = self.admin_tasks.clone();
        let health_for_run = self.health.clone();
        let health_clone = health_for_run.clone();
        let snapshot = self.state.snapshot.clone();
        let state_ref = self.state.clone();
        let rt_config = self.rt_config.clone();
        let tls_client_config = self.tls_client_config.clone();
        let compatibility_options = self.compatibility_options.clone();
        #[cfg(feature = "ssh")]
        let ssh_sessions = self.ssh_sessions.clone();

        let handshake_timeout = rt_config.timeouts.handshake;
        let connect_timeout = rt_config.timeouts.connect;

        #[cfg(feature = "operations")]
        let listener_infos_provider: Arc<RuntimeAdminListenerInfos> =
            Arc::new(RuntimeAdminListenerInfos {
                state: admin_state_ref.clone(),
            });

        #[cfg(feature = "operations")]
        let metrics_registry_for_admin = self.metrics_registry.clone();

        let run_async = async move {
            if let Ok(mut runtime) = state_ref.health_runtime.lock() {
                *runtime = Some(tokio::runtime::Handle::current());
            }
            #[cfg(feature = "operations")]
            let mut compatibility_system_proxy: Option<
                eggress_system_proxy::AppliedProxy,
            > = None;

            #[cfg(feature = "operations")]
            let metrics_registry = metrics_registry_for_admin;
            // Start health probes inside the runtime context
            {
                let mut guard = match health_for_run.lock() {
                    Ok(guard) => guard,
                    Err(error) => {
                        tracing::warn!("health manager state was poisoned; resetting it: {error}");
                        let mut guard = error.into_inner();
                        *guard = None;
                        health_for_run.clear_poison();
                        guard
                    }
                };
                if let Some(ref mut hm) = *guard {
                    let upstream_runtimes: Vec<Arc<UpstreamRuntime>> =
                        snapshot.load().upstreams.values().cloned().collect();
                    if !upstream_runtimes.is_empty() {
                        hm.start_probes(&upstream_runtimes);
                    }
                }
            }

            let current_snapshot = snapshot.load();
            let listener_configs = current_snapshot.listeners.clone();
            #[cfg(feature = "operations")]
            let admin_config = current_snapshot.admin.clone();
            drop(current_snapshot);

            if listener_configs.is_empty() {
                tracing::warn!("no listeners configured; the proxy will not accept connections");
            }

            let mut prepared = Vec::new();
            #[cfg(feature = "quic")]
            let mut prepared_quic = Vec::<PreparedQuicListener>::new();
            let compatibility_auth_reuse = compatibility_options
                .auth_timeout
                .map(eggress_server::accept::AuthReuseCache::new)
                .map(Arc::new);
            #[cfg(unix)]
            let mut unix_listener_args = Vec::new();
            let mut transparent_listener_args = Vec::new();

            for lcfg in &listener_configs {
                let protocols: Vec<ProtocolId> = lcfg.protocols.to_vec();

                let auth = match &lcfg.auth {
                    Some(auth_cfg) => {
                        if auth_cfg.auth_type == "password" {
                            let username = auth_cfg.username.clone().unwrap_or_default();
                            let password = auth_cfg.password.clone().unwrap_or_default();
                            if let Some(reuse) = compatibility_auth_reuse.clone() {
                                eggress_server::accept::InboundAuthentication::UsernamePasswordWithReuse {
                                    username,
                                    password,
                                    reuse,
                                }
                            } else {
                                eggress_server::accept::InboundAuthentication::UsernamePassword {
                                    username,
                                    password,
                                }
                            }
                        } else {
                            eggress_server::accept::InboundAuthentication::None
                        }
                    }
                    None => eggress_server::accept::InboundAuthentication::None,
                };

                let connection_limit = lcfg.connection_limit.unwrap_or(1024) as usize;

                // Handle Unix domain socket listeners separately
                #[allow(unused_variables)]
                if let Some(ref unix_cfg) = lcfg.unix {
                    #[cfg(unix)]
                    {
                        match eggress_server::listener::unix::create_unix_listener(
                            &eggress_server::listener::unix::UnixListenerConfig::from_compiled(
                                &unix_cfg.path,
                                unix_cfg.unlink_existing,
                                Some(unix_cfg.mode),
                            ),
                        ) {
                            Ok(unix_listener) => {
                                tracing::info!(
                                    "unix socket listener created at {} ({})",
                                    unix_cfg.path.display(),
                                    lcfg.name
                                );
                                unix_listener_args.push((
                                    lcfg.name.clone(),
                                    unix_listener,
                                    protocols,
                                    auth,
                                    handshake_timeout,
                                    connection_limit as u64,
                                    lcfg.tls.clone(),
                                    lcfg.shadowsocks.clone(),
                                    lcfg.trojan.clone(),
                                    lcfg.udp.clone(),
                                ));
                                continue;
                            }
                            Err(e) => {
                                tracing::error!(
                                    "failed to bind unix socket at {} for listener '{}': {e}",
                                    unix_cfg.path.display(),
                                    lcfg.name
                                );
                                continue;
                            }
                        }
                    }
                    #[cfg(not(unix))]
                    {
                        tracing::error!(
                            "unix socket listener '{}' skipped: not supported on this platform",
                            lcfg.name
                        );
                        continue;
                    }
                }

                // Handle transparent TCP listeners
                if let Some(ref transparent_cfg) = lcfg.transparent {
                    if transparent_cfg.enabled {
                        let capability = check_capability(PlatformCapability::LinuxOriginalDstIpv4);
                        if capability != crate::platform::CapabilityStatus::Available {
                            #[cfg(feature = "operations")]
                            state_ref.metrics.record_platform_capability_check_failure();
                            let _cap_span = tracing::info_span!(
                                "capability_check_failed",
                                capability = %PlatformCapability::LinuxOriginalDstIpv4,
                                status = %capability,
                                listener = %lcfg.name,
                            );
                            tracing::warn!(
                                "transparent proxy not available for listener '{}' ({}); \
                                 falling back to normal TCP listener",
                                lcfg.name,
                                capability
                            );
                        } else {
                            let bind_addr: std::net::SocketAddr =
                                lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
                                    addr: lcfg.bind.clone(),
                                    source: std::io::Error::new(
                                        std::io::ErrorKind::InvalidInput,
                                        e,
                                    ),
                                })?;

                            let transparent_listener =
                                eggress_server::listener::transparent::TransparentListener::bind(
                                    &bind_addr.to_string(),
                                )
                                .await
                                .map_err(|e| {
                                    RuntimeError::ListenerBind {
                                        addr: lcfg.bind.clone(),
                                        source: e,
                                    }
                                })?;

                            let local_addr = transparent_listener.local_addr().map_err(|e| {
                                RuntimeError::ListenerBind {
                                    addr: lcfg.bind.clone(),
                                    source: e,
                                }
                            })?;

                            tracing::info!(
                                "transparent TCP listener listening on {local_addr} ({})",
                                lcfg.name
                            );

                            transparent_listener_args.push((
                                lcfg.name.clone(),
                                transparent_listener,
                                protocols,
                                auth,
                                handshake_timeout,
                                connection_limit as u64,
                                lcfg.tls.clone(),
                                lcfg.shadowsocks.clone(),
                                lcfg.trojan.clone(),
                                lcfg.udp.clone(),
                            ));
                            continue;
                        }
                    }
                }

                #[cfg(feature = "quic")]
                if protocols.contains(&ProtocolId::Quic) || protocols.contains(&ProtocolId::Http3) {
                    let tls = lcfg.tls.clone().ok_or_else(|| {
                        RuntimeError::Other(format!(
                            "QUIC/HTTP3 listener '{}' requires certificate and key material",
                            lcfg.name
                        ))
                    })?;
                    let bind_addr: std::net::SocketAddr =
                        lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
                            addr: lcfg.bind.clone(),
                            source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
                        })?;
                    let listener = eggress_transport_quic::QuicListener::bind(
                        bind_addr,
                        eggress_transport_quic::QuicServerConfig {
                            certificate_pem: tls.cert_pem.clone(),
                            private_key_pem: tls.key_pem.clone(),
                            idle_timeout: Duration::from_secs(60),
                            max_concurrent_streams: lcfg.connection_limit.unwrap_or(1024).max(1),
                            alpn_protocols: if protocols.contains(&ProtocolId::Http3) {
                                vec![b"h3".to_vec()]
                            } else {
                                Vec::new()
                            },
                        },
                    )
                    .await
                    .map_err(|e| RuntimeError::ListenerBind {
                        addr: lcfg.bind.clone(),
                        source: std::io::Error::other(e.to_string()),
                    })?;
                    let local_addr =
                        listener
                            .local_addr()
                            .map_err(|e| RuntimeError::ListenerBind {
                                addr: lcfg.bind.clone(),
                                source: std::io::Error::other(e.to_string()),
                            })?;
                    tracing::info!("QUIC listening on {local_addr} ({})", lcfg.name);
                    prepared_quic.push(PreparedQuicListener {
                        name: lcfg.name.clone(),
                        protocols,
                        listener,
                        local_addr,
                        auth,
                        handshake_timeout,
                        connection_limit: lcfg.connection_limit.unwrap_or(1024) as u64,
                    });
                    continue;
                }

                // Standard TCP listener path
                let bind_addr: std::net::SocketAddr =
                    lcfg.bind.parse().map_err(|e| RuntimeError::ListenerBind {
                        addr: lcfg.bind.clone(),
                        source: std::io::Error::new(std::io::ErrorKind::InvalidInput, e),
                    })?;

                let config = TcpListenerConfig {
                    bind_addr,
                    protocols: protocols.clone(),
                    auth_required: false,
                    handshake_timeout,
                    connection_limit,
                };

                let listener = TcpListener::new_with_reuse_port(
                    &config,
                    listener_cancel.clone(),
                    lcfg.reuse_port.unwrap_or(false),
                )
                .await
                .map_err(|e| RuntimeError::ListenerBind {
                    addr: lcfg.bind.clone(),
                    source: e,
                })?;
                let local_addr = listener
                    .local_addr()
                    .map_err(|e| RuntimeError::ListenerBind {
                        addr: lcfg.bind.clone(),
                        source: e,
                    })?;
                tracing::info!("listening on {local_addr} ({})", lcfg.name);

                prepared.push(PreparedListener {
                    name: lcfg.name.clone(),
                    bind: lcfg.bind.clone(),
                    protocols,
                    listener,
                    local_addr,
                    auth,
                    handshake_timeout,
                    udp: lcfg.udp.clone(),
                    tls: lcfg.tls.clone(),
                    shadowsocks: lcfg.shadowsocks.clone(),
                    trojan: lcfg.trojan.clone(),
                    fixed_target: lcfg.fixed_target.clone(),
                    local_bind: lcfg.local_bind.clone(),
                });
            }

            #[cfg(feature = "operations")]
            {
                let listener_infos: Vec<eggress_admin::ListenerInfo> = prepared
                    .iter()
                    .map(|p| eggress_admin::ListenerInfo {
                        name: p.name.clone(),
                        bind: p.bind.clone(),
                        local_addr: p.local_addr.to_string(),
                        protocols: p.protocols.iter().map(|p| p.to_string()).collect(),
                        udp_enabled: p.udp.as_ref().is_some_and(|u| u.enabled),
                        mode: Some("standard".to_string()),
                        capability_status: None,
                        original_dst_support: None,
                        unix_socket_path: None,
                        unix_socket_unlink_existing: None,
                    })
                    .collect();
                drop(listener_infos);
            }

            // Store listener addresses for admin snapshot, indexed by config order.
            // Build a lookup map from all listener types (standard, transparent, unix).
            {
                let mut addr_map: std::collections::HashMap<String, Option<std::net::SocketAddr>> =
                    std::collections::HashMap::new();
                for p in &prepared {
                    addr_map.insert(p.name.clone(), Some(p.local_addr));
                }
                #[cfg(feature = "quic")]
                for p in &prepared_quic {
                    addr_map.insert(p.name.clone(), Some(p.local_addr));
                }
                for (name, transparent_listener, _, _, _, _, _, _, _, _) in
                    &transparent_listener_args
                {
                    let addr = transparent_listener.local_addr().ok();
                    addr_map.insert(name.clone(), addr);
                }
                #[cfg(unix)]
                for (name, _, _, _, _, _, _, _, _, _) in &unix_listener_args {
                    // Unix domain sockets don't have a meaningful TCP socket address
                    addr_map.insert(name.clone(), None);
                }
                let addrs: Vec<Option<std::net::SocketAddr>> = listener_configs
                    .iter()
                    .map(|lcfg| addr_map.get(&lcfg.name).copied().flatten())
                    .collect();
                let admin_addrs = addrs.clone();
                match state_ref.listener_addrs.lock() {
                    Ok(mut guard) => *guard = addrs,
                    Err(error) => {
                        tracing::warn!(
                            "listener address state was poisoned; resetting it: {error}"
                        );
                        let mut guard = error.into_inner();
                        *guard = addrs;
                        state_ref.listener_addrs.clear_poison();
                    }
                }
                #[cfg(feature = "operations")]
                state_ref.publish_admin_listener_addrs(state_ref.snapshot.load_full(), admin_addrs);
            }

            #[cfg(feature = "extended")]
            let mut shadowsocks_udp_relays = Vec::new();
            let mut echo_udp_relays = Vec::new();

            for prepared_listener in &prepared {
                if let Some(ref udp_cfg) = prepared_listener.udp {
                    #[cfg(feature = "extended")]
                    if udp_cfg.mode == eggress_udp::UdpMode::ShadowsocksUdp {
                        shadowsocks_udp_relays.push(
                            prepare_shadowsocks_udp_relay(
                                prepared_listener,
                                udp_cfg,
                                routing.clone(),
                                &state_ref,
                            )
                            .await?,
                        );
                    }
                    if udp_cfg.mode == eggress_udp::UdpMode::Echo {
                        let socket =
                            Arc::new(tokio::net::UdpSocket::bind(udp_cfg.bind).await.map_err(
                                |e| RuntimeError::ListenerBind {
                                    addr: udp_cfg.bind.to_string(),
                                    source: e,
                                },
                            )?);
                        echo_udp_relays.push((prepared_listener.name.clone(), socket, None));
                    } else if udp_cfg.mode == eggress_udp::UdpMode::FixedTarget {
                        let socket =
                            Arc::new(tokio::net::UdpSocket::bind(udp_cfg.bind).await.map_err(
                                |e| RuntimeError::ListenerBind {
                                    addr: udp_cfg.bind.to_string(),
                                    source: e,
                                },
                            )?);
                        echo_udp_relays.push((
                            prepared_listener.name.clone(),
                            socket,
                            udp_cfg.fixed_target.clone(),
                        ));
                    }
                }
            }

            #[cfg(feature = "extended")]
            for (socket, relay_config) in shadowsocks_udp_relays {
                let relay_cancel = cancel.clone();
                tasks.spawn(async move {
                    let result =
                        eggress_udp::standalone_shadowsocks::shadowsocks_standalone_udp_relay(
                            socket,
                            relay_config,
                            relay_cancel,
                        )
                        .await;
                    if let Err(error) = result {
                        tracing::debug!(
                            %error,
                            "Shadowsocks UDP relay ended with error"
                        );
                    }
                });
            }

            for (listener_name, socket, fixed_target) in echo_udp_relays {
                let relay_cancel = cancel.clone();
                let max_size = listener_configs
                    .iter()
                    .find(|l| l.name == listener_name)
                    .and_then(|l| l.udp.as_ref())
                    .map(|u| u.max_datagram_size)
                    .unwrap_or(65535);
                tasks.spawn(async move {
                    let mut buf = vec![0u8; max_size];
                    const ECHO_DNS_TIMEOUT: Duration = Duration::from_secs(2);
                    const ECHO_CONNECT_TIMEOUT: Duration = Duration::from_secs(2);
                    const ECHO_RECV_TIMEOUT: Duration = Duration::from_secs(5);
                    let target_socket = if let Some(target) = fixed_target {
                        let lookup = async {
                            match target.host {
                                eggress_core::TargetHost::Ip(ip) => {
                                    Ok(std::net::SocketAddr::new(ip, target.port))
                                }
                                eggress_core::TargetHost::Domain(domain) => match tokio::net::lookup_host((domain.as_str(), target.port)).await {
                                    Ok(mut addrs) => match addrs.next() {
                                        Some(addr) => Ok(addr),
                                        None => Err(()),
                                    },
                                    Err(_) => Err(()),
                                },
                            }
                        };
                        let addr = match tokio::time::timeout(ECHO_DNS_TIMEOUT, lookup).await {
                            Ok(Ok(a)) => a,
                            Ok(Err(())) | Err(_) => return,
                        };
                        let target_socket = match tokio::net::UdpSocket::bind("0.0.0.0:0").await {
                            Ok(s) => s,
                            Err(_) => return,
                        };
                        if tokio::time::timeout(ECHO_CONNECT_TIMEOUT, target_socket.connect(addr))
                            .await
                            .map(|r| r.is_ok())
                            .unwrap_or(false)
                        {
                            Some(target_socket)
                        } else {
                            return;
                        }
                    } else { None };
                    let mut response = vec![0u8; max_size];
                    loop {
                        tokio::select! {
                            result = socket.recv_from(&mut buf) => match result {
                                Ok((n, peer)) => {
                                    if let Some(ref target_socket) = target_socket {
                                        if target_socket.send(&buf[..n]).await.is_ok() {
                                            if let Ok(Ok(m)) = tokio::time::timeout(ECHO_RECV_TIMEOUT, target_socket.recv(&mut response)).await {
                                                let _ = socket.send_to(&response[..m], peer).await;
                                            }
                                        }
                                    } else { let _ = socket.send_to(&buf[..n], peer).await; }
                                }
                                Err(_) => break,
                            },
                            _ = relay_cancel.cancelled() => break,
                        }
                    }
                });
            }

            // Spawn transparent listener accept loops
            for (
                listener_name,
                transparent_listener,
                protocols,
                auth,
                hs_timeout,
                connection_limit,
                tls_cfg,
                ss_cfg,
                trojan_cfg,
                udp_cfg,
            ) in transparent_listener_args
            {
                let routing = routing.clone();
                let state = state_ref.clone();
                let conn_tasks = connection_tasks.clone();
                let conn_cancel = connection_cancel.clone();
                let tls_client_config = tls_client_config.clone();
                let listener_cancel = listener_cancel.clone();

                #[cfg(feature = "ssh")]
                let listener_ssh_sessions = ssh_sessions.clone();

                tasks.spawn(async move {
                    let proto_slice: Arc<[ProtocolId]> = protocols.clone().into();
                    let transparent_listener_inner = transparent_listener.inner();
                    let listener_active = Arc::new(AtomicU64::new(0));

                    let transparent_accepted = state.transparent_accepted_total.clone();
                    let transparent_dst_failed =
                        state.transparent_original_dst_failed_total.clone();

                    loop {
                        let accept_result = tokio::select! {
                            result = transparent_listener_inner.accept() => result,
                            _ = listener_cancel.cancelled() => {
                                break;
                            }
                        };

                        let (stream, _peer) = match accept_result {
                            Ok(s) => s,
                            Err(e) => {
                                // Cancellation is handled by the token select
                                // above; the inner accept cannot report it.
                                handle_accept_error(
                                    &format!("transparent on '{listener_name}'"),
                                    &e,
                                )
                                .await;
                                continue;
                            }
                        };

                        let connection_slot = match ListenerConnectionSlot::try_acquire(
                            &listener_active,
                            connection_limit,
                        ) {
                            Some(slot) => slot,
                            None => {
                                tracing::debug!(
                                    listener = %listener_name,
                                    limit = connection_limit,
                                    "dropping transparent connection: connection limit reached"
                                );
                                drop(stream);
                                continue;
                            }
                        };

                        transparent_accepted.fetch_add(1, Ordering::Relaxed);

                        let original_dst =
                            match eggress_server::listener::transparent::get_original_destination(&stream) {
                                Ok(addr) => addr,
                                Err(e) => {
                                    drop(connection_slot);
                                    transparent_dst_failed.fetch_add(1, Ordering::Relaxed);
                                    let _span = tracing::info_span!(
                                        "transparent_original_dst_failed",
                                        listener = %listener_name,
                                        error = %e,
                                    );
                                    tracing::warn!(
                                        "failed to get original destination for transparent connection on '{}': {e}",
                                        listener_name
                                    );
                                    continue;
                                }
                            };

                        let peer = stream
                            .peer_addr()
                            .unwrap_or_else(|_| {
                                std::net::SocketAddr::new(
                                    std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED),
                                    0,
                                )
                            });

                        let routing = routing.clone();
                        let tls_client_config = tls_client_config.clone();
                        let listener_str = listener_name.clone();
                        let conn_id = state
                            .connection_counter
                            .fetch_add(1, Ordering::Relaxed);
                        let conn_protocols = proto_slice.clone();
                        let conn_auth = auth.clone();
                        let conn_metrics = state.metrics.clone();
                        #[cfg(feature = "extended")]
                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
                        let active = state.active_connections.clone();
                        let conn_cancel = conn_cancel.child_token();
                        let generation = state.snapshot.load().generation;
                        let tls_config = tls_cfg.clone();
                        let ss_config = ss_cfg.clone();
                        let trojan_config = trojan_cfg.clone();

                        let udp_svc = udp_cfg.as_ref().map(|udp_config| {
                            Arc::new(RuntimeUdpService {
                                _listener_name: listener_name.clone(),
                                udp_config: udp_config.clone(),
                                registry: state.udp_registry.clone(),
                                metrics: state.metrics.clone(),
                                udp_metrics: state.udp_metrics.clone(),
                                routing: routing.clone(),
                                udp_tasks: state.udp_tasks.clone(),
                            }) as Arc<dyn eggress_server::UdpService>
                        });

                        #[cfg(feature = "ssh")]
                        let conn_ssh_sessions = listener_ssh_sessions.clone();
                        conn_tasks.spawn(async move {
                            let _active_guard = ActiveConnectionGuard::new(active);
                            let _connection_slot = connection_slot;
                            let started = std::time::Instant::now();

                            let stream: eggress_core::BoxStream =
                                if let Some(ref tls_cfg) = tls_config {
                                    let server_config = match eggress_transport_tls::TlsServerConfigBuilder::new()
                                        .with_certificate_pem(&tls_cfg.cert_pem)
                                        .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
                                        .and_then(|b| {
                                            let b = if tls_cfg.alpn.is_empty() { b } else { b.with_alpn(tls_cfg.alpn.clone()) };
                                            b.build()
                                        }) {
                                            Ok(c) => c,
                                            Err(e) => {
                                                tracing::error!(%peer, "TLS config error: {e}");
                                                return;
                                            }
                                        };
                                    match eggress_transport_tls::tls_accept(Box::new(stream), server_config).await {
                                        Ok(s) => s,
                                        Err(e) => {
                                            tracing::debug!(%peer, "TLS accept failed: {e}");
                                            return;
                                        }
                                    }
                                } else {
                                    Box::new(stream)
                                };

                            let config = eggress_server::ConnectionConfig {
                                routing: routing as Arc<dyn RouteService>,
                                context: eggress_server::ConnectionContext {
                                    source: Some(peer),
                                    listener: listener_str.clone(),
                                    generation,
                                },
                                handshake_timeout: hs_timeout,
                                connect_timeout,
                                protocols: conn_protocols,
                                authentication: conn_auth,
                                metrics: Some(conn_metrics),
                                udp: udp_svc,
                                tls_client_config,
                                shadowsocks: ss_config.map(
                                    |ss| eggress_server::accept::InboundShadowsocksConfig {
                                        method: ss.method.clone(),
                                        password: ss.password.clone(),
                                        #[cfg(feature = "pproxy-legacy")]
                                        auth_prefix: ss.auth_prefix.clone().map(String::into_bytes),
                                        #[cfg(feature = "pproxy-legacy")]
                                        plugins: ss.plugins.clone(),
                                    },
                                ),
                                #[cfg(feature = "extended")]
                                shadowsocks_metrics: Some(conn_ss_metrics),
                                #[cfg(not(feature = "extended"))]
                                shadowsocks_metrics: None,
                                trojan: trojan_config.map(
                                    |t| eggress_server::accept::InboundTrojanConfig {
                                        password: t.password.clone(),
                                        fallback: t.fallback.clone(),
                                    },
                                ),
                                fixed_target: None,
                                local_bind: None,
                                #[cfg(feature = "ssh")]
                                ssh_sessions: Some(conn_ssh_sessions),
                            };

                            let report = tokio::select! {
                                report = eggress_server::serve_connection(stream, config)
                                    .instrument(tracing::info_span!(
                                        "conn",
                                        id = conn_id,
                                        peer = %peer,
                                        original_dst = %original_dst,
                                        listener_type = "transparent",
                                        listener = %listener_str,
                                    )) => {
                                report
                            }
                                _ = conn_cancel.cancelled() => {
                                    eggress_server::SessionReport::cancelled(
                                        None,
                                        None,
                                        String::new(),
                                    )
                                }
                            };

                            tracing::info!(
                                protocol = ?report.protocol,
                                target = ?report.target,
                                original_dst = %original_dst,
                                route = %report.route,
                                outcome = ?report.outcome,
                                bytes_upstream = report.bytes_upstream,
                                bytes_downstream = report.bytes_downstream,
                                duration_ms = started.elapsed().as_millis() as u64,
                                "transparent connection completed",
                            );
                        });
                    }
                });
            }

            // Spawn Unix domain socket accept loops
            #[cfg(unix)]
            for (
                listener_name,
                unix_listener,
                protocols,
                auth,
                hs_timeout,
                connection_limit,
                tls_cfg,
                ss_cfg,
                trojan_cfg,
                udp_cfg,
            ) in unix_listener_args
            {
                let routing = routing.clone();
                let state = state_ref.clone();
                let conn_tasks = connection_tasks.clone();
                let conn_cancel = connection_cancel.clone();
                let tls_client_config = tls_client_config.clone();
                let listener_cancel = listener_cancel.clone();

                let socket_path = unix_listener.path().display().to_string();

                #[cfg(feature = "ssh")]
                let listener_ssh_sessions = ssh_sessions.clone();
                tasks.spawn(async move {
                    let proto_slice: Arc<[ProtocolId]> = protocols.clone().into();
                    let listener_active = Arc::new(AtomicU64::new(0));

                    let _accept_loop_span = tracing::info_span!(
                        "unix_accept_loop",
                        listener = %listener_name,
                        socket_path = %socket_path,
                    );

                    loop {
                        let (stream, _peer_addr) = tokio::select! {
                            result = unix_listener.accept() => match result {
                                Ok(r) => r,
                                Err(e) => {
                                    handle_accept_error("unix", &e).await;
                                    continue;
                                }
                            },
                            _ = listener_cancel.cancelled() => {
                                break;
                            }
                        };

                        let connection_slot = match ListenerConnectionSlot::try_acquire(
                            &listener_active,
                            connection_limit,
                        ) {
                            Some(slot) => slot,
                            None => {
                                tracing::debug!(
                                    listener = %listener_name,
                                    limit = connection_limit,
                                    "dropping Unix connection: connection limit reached"
                                );
                                drop(stream);
                                continue;
                            }
                        };

                        state.metrics.record_unix_listener_connection_accepted();

                        let routing = routing.clone();
                        let tls_client_config = tls_client_config.clone();
                        let listener_str = listener_name.clone();
                        let conn_id = state.connection_counter.fetch_add(1, Ordering::Relaxed);
                        let conn_protocols = proto_slice.clone();
                        let conn_auth = auth.clone();
                        let conn_metrics = state.metrics.clone();
                        #[cfg(feature = "extended")]
                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
                        let active = state.active_connections.clone();
                        let conn_cancel = conn_cancel.child_token();
                        let generation = state.snapshot.load().generation;

                        let tls_config = tls_cfg.clone();
                        let ss_config = ss_cfg.clone();
                        let trojan_config = trojan_cfg.clone();
                        let socket_path_clone = socket_path.clone();
                        let listener_str_for_span = listener_str.clone();

                        let udp_svc = udp_cfg.as_ref().map(|udp_config| {
                            Arc::new(RuntimeUdpService {
                                _listener_name: listener_name.clone(),
                                udp_config: udp_config.clone(),
                                registry: state.udp_registry.clone(),
                                metrics: state.metrics.clone(),
                                udp_metrics: state.udp_metrics.clone(),
                                routing: routing.clone(),
                                udp_tasks: state.udp_tasks.clone(),
                            }) as Arc<dyn eggress_server::UdpService>
                        });

                        #[cfg(feature = "ssh")]
                        let conn_ssh_sessions = listener_ssh_sessions.clone();
                        conn_tasks.spawn(async move {
                            let _active_guard = ActiveConnectionGuard::new(active);
                            let _connection_slot = connection_slot;
                            let started = std::time::Instant::now();

                            let stream: eggress_core::BoxStream =
                                if let Some(ref tls_cfg) = tls_config {
                                    let server_config =
                                        match eggress_transport_tls::TlsServerConfigBuilder::new()
                                            .with_certificate_pem(&tls_cfg.cert_pem)
                                            .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
                                            .and_then(|b| {
                                                let b = if tls_cfg.alpn.is_empty() {
                                                    b
                                                } else {
                                                    b.with_alpn(tls_cfg.alpn.clone())
                                                };
                                                b.build()
                                            }) {
                                            Ok(c) => c,
                                            Err(e) => {
                                                tracing::error!(
                                                    "TLS config error for unix connection: {e}"
                                                );
                                                return;
                                            }
                                        };
                                    match eggress_transport_tls::tls_accept(
                                        Box::new(stream),
                                        server_config,
                                    )
                                    .await
                                    {
                                        Ok(s) => s,
                                        Err(e) => {
                                            tracing::debug!(
                                                "TLS accept failed for unix connection: {e}"
                                            );
                                            return;
                                        }
                                    }
                                } else {
                                    Box::new(stream)
                                };

                            let peer = std::net::SocketAddr::new(
                                std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
                                0,
                            );

                            let config = eggress_server::ConnectionConfig {
                                routing: routing as Arc<dyn RouteService>,
                                context: eggress_server::ConnectionContext {
                                    source: Some(peer),
                                    listener: listener_str,
                                    generation,
                                },
                                handshake_timeout: hs_timeout,
                                connect_timeout,
                                protocols: conn_protocols,
                                authentication: conn_auth,
                                metrics: Some(conn_metrics),
                                udp: udp_svc,
                                tls_client_config,
                                shadowsocks: ss_config.map(|ss| {
                                    eggress_server::accept::InboundShadowsocksConfig {
                                        method: ss.method.clone(),
                                        password: ss.password.clone(),
                                        #[cfg(feature = "pproxy-legacy")]
                                        auth_prefix: ss.auth_prefix.clone().map(String::into_bytes),
                                        #[cfg(feature = "pproxy-legacy")]
                                        plugins: ss.plugins.clone(),
                                    }
                                }),
                                #[cfg(feature = "extended")]
                                shadowsocks_metrics: Some(conn_ss_metrics),
                                #[cfg(not(feature = "extended"))]
                                shadowsocks_metrics: None,
                                trojan: trojan_config.map(|t| {
                                    eggress_server::accept::InboundTrojanConfig {
                                        password: t.password.clone(),
                                        fallback: t.fallback.clone(),
                                    }
                                }),
                                fixed_target: None,
                                local_bind: None,
                                #[cfg(feature = "ssh")]
                                ssh_sessions: Some(conn_ssh_sessions),
                            };

                            let report = tokio::select! {
                                report = eggress_server::serve_connection(stream, config)
                                    .instrument(tracing::info_span!(
                                        "conn",
                                        id = conn_id,
                                        peer = %peer,
                                        listener_type = "unix",
                                        listener = %listener_str_for_span,
                                        socket_path = %socket_path_clone,
                                    )) => {
                                    report
                                }
                                _ = conn_cancel.cancelled() => {
                                    eggress_server::SessionReport::cancelled(
                                        None,
                                        None,
                                        String::new(),
                                    )
                                }
                            };

                            tracing::info!(
                                protocol = ?report.protocol,
                                target = ?report.target,
                                route = %report.route,
                                outcome = ?report.outcome,
                                bytes_upstream = report.bytes_upstream,
                                bytes_downstream = report.bytes_downstream,
                                duration_ms = started.elapsed().as_millis() as u64,
                                "unix connection completed",
                            );
                        });
                    }

                    // Cleanup socket file on shutdown
                    unix_listener.cleanup().unwrap_or_else(|e| {
                        tracing::warn!("failed to cleanup unix socket: {e}");
                    });
                });
            }

            // Spawn standard TCP accept loops
            #[cfg(feature = "operations")]
            let compatibility_proxy_selection = prepared
                .iter()
                .find(|listener| {
                    listener.protocols.contains(&ProtocolId::Socks5)
                        && listener.local_addr.port() != 0
                })
                .map(|listener| {
                    (
                        eggress_system_proxy::CompatibilityProxyKind::Socks5,
                        listener.local_addr.port(),
                    )
                })
                .or_else(|| {
                    prepared
                        .iter()
                        .find(|listener| {
                            listener.protocols.contains(&ProtocolId::Http)
                                && listener.local_addr.port() != 0
                        })
                        .map(|listener| {
                            (
                                eggress_system_proxy::CompatibilityProxyKind::Http,
                                listener.local_addr.port(),
                            )
                        })
                });

            // All compatibility TCP listeners are bound before this point,
            // so --sys can use the actual selected port. Apply before any
            // accept loop or admin task is started; an apply failure is
            // therefore still a pre-run startup error.
            if compatibility_options.system_proxy {
                #[cfg(feature = "operations")]
                {
                    let selected = compatibility_proxy_selection.ok_or_else(|| {
                        RuntimeError::Other(
                            "--sys requires a usable local HTTP or SOCKS5 listener".to_string(),
                        )
                    })?;
                    let address = std::net::SocketAddr::new(
                        std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
                        selected.1,
                    );
                    compatibility_system_proxy = Some(
                        eggress_system_proxy::apply_compatibility_proxy(selected.0, address)
                            .map_err(RuntimeError::Other)?,
                    );
                }
                #[cfg(not(feature = "operations"))]
                {
                    return Err(RuntimeError::Other(
                        "--sys requires the operations feature".to_string(),
                    ));
                }
            }

            #[cfg(feature = "quic")]
            for prepared_listener in prepared_quic {
                let listener_name = prepared_listener.name.clone();
                let listener_protocols: Arc<[ProtocolId]> = prepared_listener
                    .protocols
                    .iter()
                    .copied()
                    .filter(|protocol| !matches!(protocol, ProtocolId::Quic | ProtocolId::Http3))
                    .collect::<Vec<_>>()
                    .into();
                let routing = routing.clone();
                let state = state_ref.clone();
                let conn_tasks = connection_tasks.clone();
                let conn_cancel = connection_cancel.clone();
                let listener_cancel = listener_cancel.clone();
                let is_h3 = prepared_listener.protocols.contains(&ProtocolId::Http3);
                let auth = prepared_listener.auth.clone();
                let handshake_timeout_for_listener = prepared_listener.handshake_timeout;
                let connection_limit = prepared_listener.connection_limit;

                #[cfg(feature = "ssh")]
                let listener_ssh_sessions = ssh_sessions.clone();

                if is_h3 {
                    let listener = prepared_listener.listener.clone();
                    let tls_client_config_for_listener = tls_client_config.clone();
                    tasks.spawn(async move {
                        let active_streams = Arc::new(AtomicU64::new(0));
                        loop {
                            let connection = match listener.accept_connection(&listener_cancel).await {
                                Ok(Some(connection)) => connection,
                                Ok(None) => break,
                                Err(error) => {
                                    tracing::debug!(%error, listener = %listener_name, "H3 connection failed");
                                    continue;
                                }
                            };
                            let routing = routing.clone();
                            let state = state.clone();
                            let conn_tasks = conn_tasks.clone();
                            let conn_cancel = conn_cancel.child_token();
                            let listener_name = listener_name.clone();
                            let auth = auth.clone();
                            let authorization = match &auth {
                                eggress_server::accept::InboundAuthentication::None => None,
                                eggress_server::accept::InboundAuthentication::UsernamePassword { username, password }
                                | eggress_server::accept::InboundAuthentication::UsernamePasswordWithReuse { username, password, .. } => {
                                    Some((username.clone(), password.clone()))
                                }
                            };
                            let active_streams = active_streams.clone();
                            let protocols = listener_protocols.clone();
                            let connection_limit = connection_limit;
                            let tls_client_config_for_connection = tls_client_config_for_listener.clone();
                            #[cfg(feature = "ssh")]
                            let listener_ssh_sessions_for_connection = listener_ssh_sessions.clone();
                            conn_tasks.spawn(async move {
                                let result = eggress_protocol_h3::serve_connection(
                                    connection,
                                    conn_cancel.clone(),
                                    authorization,
                                    move |request, stream, peer| {
                                        let routing = routing.clone();
                                        let state = state.clone();
                                        let listener_name = listener_name.clone();
                                        let auth = auth.clone();
                                        let protocols = protocols.clone();
                                        let active_streams = active_streams.clone();
                                        let tls_client_config = tls_client_config_for_connection.clone();
                                        #[cfg(feature = "ssh")]
                                        let ssh_sessions = listener_ssh_sessions_for_connection.clone();
                                        async move {
                                            let slot = match ListenerConnectionSlot::try_acquire(&active_streams, connection_limit) {
                                                Some(slot) => slot,
                                                None => return,
                                            };
                                            let target = match request.target() {
                                                Ok(target) => target,
                                                Err(error) => {
                                                    tracing::debug!(%error, "invalid H3 CONNECT authority");
                                                    drop(slot);
                                                    return;
                                                }
                                            };
                                            let generation = state.snapshot.load().generation;
                                            let config = eggress_server::ConnectionConfig {
                                                routing: routing as Arc<dyn RouteService>,
                                                context: eggress_server::ConnectionContext {
                                                    source: Some(peer),
                                                    listener: listener_name,
                                                    generation,
                                                },
                                                handshake_timeout: handshake_timeout_for_listener,
                                                connect_timeout,
                                                protocols,
                                                authentication: auth,
                                                metrics: Some(state.metrics.clone()),
                                                udp: None,
                                                tls_client_config,
                                                shadowsocks: None,
                                                #[cfg(feature = "extended")]
                                                shadowsocks_metrics: Some(state.shadowsocks_metrics.clone()),
                                                #[cfg(not(feature = "extended"))]
                                                shadowsocks_metrics: None,
                                                trojan: None,
                                                fixed_target: None,
                                                local_bind: None,
                                                #[cfg(feature = "ssh")]
                                                ssh_sessions: Some(ssh_sessions),
                                            };
                                            let pending = eggress_server::accept::PendingTunnel {
                                                target: target.clone(),
                                                client: stream,
                                                protocol: eggress_server::accept::TunnelProtocol::Http3,
                                                reply_context: eggress_server::accept::ReplyContext::Http3,
                                                identity: eggress_core::ClientIdentity::Anonymous,
                                            };
                                            state.metrics.record_session_start();
                                            let report = eggress_server::execute::execute(
                                                eggress_server::accept::AcceptedSession::Tunnel(pending),
                                                &config,
                                            ).await;
                                            state.metrics.record_session(&report);
                                            drop(slot);
                                        }
                                    },
                                ).await;
                                if let Err(error) = result {
                                    tracing::debug!(%error, "H3 connection ended");
                                }
                            });
                        }
                    });
                } else {
                    let listener = prepared_listener.listener.clone();
                    let auth = prepared_listener.auth.clone();
                    let tls_client_config_for_listener = tls_client_config.clone();
                    tasks.spawn(async move {
                        let active_streams = Arc::new(AtomicU64::new(0));
                        let listener_name_for_handler = listener_name.clone();
                        let protocols_for_handler = listener_protocols.clone();
                        let routing_for_handler = routing.clone();
                        let state_for_handler = state.clone();
                        let auth_for_handler = auth.clone();
                        #[cfg(feature = "ssh")]
                        let listener_ssh_sessions_for_handler = Some(listener_ssh_sessions.clone());
                        let result = listener
                            .run(listener_cancel, move |stream, peer| {
                                let routing = routing_for_handler.clone();
                                let state = state_for_handler.clone();
                                let listener_name = listener_name_for_handler.clone();
                                let protocols = protocols_for_handler.clone();
                                let auth = auth_for_handler.clone();
                                let active_streams = active_streams.clone();
                                #[cfg(feature = "ssh")]
                                let ssh_sessions = listener_ssh_sessions_for_handler.clone();
                                let tls_client_config = tls_client_config_for_listener.clone();
                                async move {
                                    let Some(slot) = ListenerConnectionSlot::try_acquire(
                                        &active_streams,
                                        connection_limit,
                                    ) else {
                                        return;
                                    };
                                    let generation = state.snapshot.load().generation;
                                    let config = eggress_server::ConnectionConfig {
                                        routing: routing as Arc<dyn RouteService>,
                                        context: eggress_server::ConnectionContext {
                                            source: Some(peer),
                                            listener: listener_name,
                                            generation,
                                        },
                                        handshake_timeout: handshake_timeout_for_listener,
                                        connect_timeout,
                                        protocols,
                                        authentication: auth,
                                        metrics: Some(state.metrics.clone()),
                                        udp: None,
                                        tls_client_config,
                                        shadowsocks: None,
                                        #[cfg(feature = "extended")]
                                        shadowsocks_metrics: Some(
                                            state.shadowsocks_metrics.clone(),
                                        ),
                                        #[cfg(not(feature = "extended"))]
                                        shadowsocks_metrics: None,
                                        trojan: None,
                                        fixed_target: None,
                                        local_bind: None,
                                        #[cfg(feature = "ssh")]
                                        ssh_sessions,
                                    };
                                    let _ = eggress_server::serve_connection(stream, config).await;
                                    drop(slot);
                                }
                            })
                            .await;
                        if let Err(error) = result {
                            tracing::debug!(%error, "QUIC listener ended");
                        }
                    });
                }
            }

            for prepared_listener in prepared {
                let routing = routing.clone();
                let state = state_ref.clone();
                let conn_tasks = connection_tasks.clone();
                let conn_cancel = connection_cancel.clone();
                let tls_client_config = tls_client_config.clone();

                #[cfg(feature = "ssh")]
                let listener_ssh_sessions = ssh_sessions.clone();
                tasks.spawn(async move {
                    let proto_slice: Arc<[ProtocolId]> = prepared_listener.protocols.clone().into();

                    loop {
                        let conn = match prepared_listener.listener.accept().await {
                            Ok(c) => c,
                            Err(e) => {
                                if is_listener_cancelled(&e) {
                                    break;
                                }
                                handle_accept_error("tcp", &e).await;
                                continue;
                            }
                        };

                        let routing = routing.clone();
                        let tls_client_config = tls_client_config.clone();
                        let peer = conn.peer_addr;
                        let listener_str = prepared_listener.name.clone();
                        let conn_id = state.connection_counter.fetch_add(1, Ordering::Relaxed);
                        let conn_protocols = proto_slice.clone();
                        let conn_auth = prepared_listener.auth.clone();
                        let conn_metrics = state.metrics.clone();
                        #[cfg(feature = "extended")]
                        let conn_ss_metrics = state.shadowsocks_metrics.clone();
                        let active = state.active_connections.clone();
                        let conn_cancel = conn_cancel.child_token();
                        let generation = state.snapshot.load().generation;

                        let tls_config = prepared_listener.tls.clone();
                        let ss_config = prepared_listener.shadowsocks.clone();
                        let trojan_config = prepared_listener.trojan.clone();
                        let fixed_target = prepared_listener.fixed_target.clone();
                        let local_bind = prepared_listener.local_bind.clone();

                        let udp_svc = if let Some(ref udp_config) = prepared_listener.udp {
                            Some(Arc::new(RuntimeUdpService {
                                _listener_name: prepared_listener.name.clone(),
                                udp_config: udp_config.clone(),
                                registry: state.udp_registry.clone(),
                                metrics: state.metrics.clone(),
                                udp_metrics: state.udp_metrics.clone(),
                                routing: routing.clone(),
                                udp_tasks: state.udp_tasks.clone(),
                            })
                                as Arc<dyn eggress_server::UdpService>)
                        } else {
                            None
                        };
                        #[cfg(feature = "ssh")]
                        let conn_ssh_sessions = listener_ssh_sessions.clone();
                        let stream_tasks = conn_tasks.clone();
                        conn_tasks.spawn(async move {
                            let _active_guard = ActiveConnectionGuard::new(active);
                            let started = std::time::Instant::now();

                            // Apply TLS if configured for this listener
                            let stream: eggress_core::BoxStream =
                                if let Some(ref tls_cfg) = tls_config {
                                    let server_config =
                                        match eggress_transport_tls::TlsServerConfigBuilder::new()
                                            .with_certificate_pem(&tls_cfg.cert_pem)
                                            .and_then(|b| b.with_key_pem(&tls_cfg.key_pem))
                                            .and_then(|b| {
                                                let b = if tls_cfg.alpn.is_empty() {
                                                    b
                                                } else {
                                                    b.with_alpn(tls_cfg.alpn.clone())
                                                };
                                                b.build()
                                            }) {
                                            Ok(c) => c,
                                            Err(e) => {
                                                tracing::error!(%peer, "TLS config error: {e}");
                                                return;
                                            }
                                        };
                                    match eggress_transport_tls::tls_accept(
                                        Box::new(conn.stream),
                                        server_config,
                                    )
                                    .await
                                    {
                                        Ok(s) => s,
                                        Err(e) => {
                                            tracing::debug!(%peer, "TLS accept failed: {e}");
                                            return;
                                        }
                                    }
                                } else {
                                    Box::new(conn.stream)
                                };

                            #[cfg(feature = "extended")]
                            let advanced_protocol = conn_protocols.first().copied();
                            #[cfg(feature = "extended")]
                            let advanced_is_single = conn_protocols.len() == 1;
                            #[cfg(feature = "extended")]
                            let advanced_fixed_target = fixed_target.clone();
                            let config = eggress_server::ConnectionConfig {
                                routing: routing as Arc<dyn RouteService>,
                                context: eggress_server::ConnectionContext {
                                    source: Some(peer),
                                    listener: listener_str,
                                    generation,
                                },
                                handshake_timeout: prepared_listener.handshake_timeout,
                                connect_timeout,
                                protocols: conn_protocols,
                                authentication: conn_auth,
                                metrics: Some(conn_metrics),
                                udp: udp_svc,
                                tls_client_config: tls_client_config.clone(),
                                shadowsocks: ss_config.map(|ss| {
                                    eggress_server::accept::InboundShadowsocksConfig {
                                        method: ss.method.clone(),
                                        password: ss.password.clone(),
                                        #[cfg(feature = "pproxy-legacy")]
                                        auth_prefix: ss.auth_prefix.clone().map(String::into_bytes),
                                        #[cfg(feature = "pproxy-legacy")]
                                        plugins: ss.plugins.clone(),
                                    }
                                }),
                                #[cfg(feature = "extended")]
                                shadowsocks_metrics: Some(conn_ss_metrics),
                                #[cfg(not(feature = "extended"))]
                                shadowsocks_metrics: None,
                                trojan: trojan_config.map(|t| {
                                    eggress_server::accept::InboundTrojanConfig {
                                        password: t.password.clone(),
                                        fallback: t.fallback.clone(),
                                    }
                                }),
                                fixed_target,
                                local_bind,
                                #[cfg(feature = "ssh")]
                                ssh_sessions: Some(conn_ssh_sessions),
                            };

                            #[cfg(feature = "extended")]
                            if matches!(
                                advanced_protocol,
                                Some(ProtocolId::Http2 | ProtocolId::WebSocket)
                            ) && advanced_is_single
                            {
                                let advanced_result = match advanced_protocol {
                                    Some(ProtocolId::Http2) => {
                                        eggress_server::advanced::serve_h2_connection(
                                            stream,
                                            config,
                                            &stream_tasks,
                                            conn_cancel.clone(),
                                        )
                                        .await
                                    }
                                    Some(ProtocolId::WebSocket) => match advanced_fixed_target {
                                        Some(target) => {
                                            eggress_server::advanced::serve_websocket_connection(
                                                stream, config, target,
                                            )
                                            .await
                                        }
                                        None => Err("WebSocket listener requires a fixed target"
                                            .to_string()),
                                    },
                                    None => {
                                        Err("advanced listener has no configured protocol"
                                            .to_string())
                                    }
                                    Some(_) => {
                                        Err("advanced listener protocol is not supported here"
                                            .to_string())
                                    }
                                };
                                if let Err(error) = advanced_result {
                                    tracing::debug!(%peer, %error, "advanced listener ended");
                                }
                                return;
                            }

                            let report = tokio::select! {
                                report = eggress_server::serve_connection(stream, config)
                                    .instrument(tracing::info_span!(
                                        "conn",
                                        id = conn_id,
                                        peer = %peer,
                                    )) => {
                                    report
                                }
                                _ = conn_cancel.cancelled() => {
                                    eggress_server::SessionReport::cancelled(
                                        None,
                                        None,
                                        String::new(),
                                    )
                                }
                            };

                            if compatibility_options.debug && report.failure.is_some() {
                                tracing::error!(
                                    protocol = ?report.protocol,
                                    target = ?report.target,
                                    route = %report.route,
                                    outcome = ?report.outcome,
                                    failure = ?report.failure,
                                    "pproxy debug connection failure",
                                );
                            } else {
                                tracing::info!(
                                    protocol = ?report.protocol,
                                    target = ?report.target,
                                    route = %report.route,
                                    outcome = ?report.outcome,
                                    bytes_upstream = report.bytes_upstream,
                                    bytes_downstream = report.bytes_downstream,
                                    duration_ms = started.elapsed().as_millis() as u64,
                                    "connection completed",
                                );
                            }
                            if compatibility_options.verbose_level >= 1 {
                                tracing::info!(
                                    protocol = ?report.protocol,
                                    target = ?report.target,
                                    route = %report.route,
                                    "pproxy connection event",
                                );
                            }
                            if compatibility_options.verbose_level >= 2 {
                                tracing::info!(
                                    bytes_upstream = report.bytes_upstream,
                                    bytes_downstream = report.bytes_downstream,
                                    "pproxy traffic stats",
                                );
                            }
                        });
                    }
                });
            }

            // Spawn reverse servers and clients
            #[cfg(feature = "reverse")]
            {
                let current_snapshot = snapshot.load();
                let reverse_servers = current_snapshot.reverse_servers.clone();
                let reverse_clients = current_snapshot.reverse_clients.clone();
                drop(current_snapshot);

                for rs_cfg in reverse_servers {
                    if rs_cfg.pproxy_compat {
                        let server_config =
                            eggress_protocol_reverse::compat_pproxy::PproxyBackwardServerConfig {
                                control_bind: rs_cfg.control_bind,
                                external_bind: rs_cfg.external_bind,
                                auth: eggress_protocol_reverse::compat_pproxy::raw_auth(
                                    rs_cfg.auth_username.as_deref(),
                                    rs_cfg.auth_password.as_deref(),
                                ),
                                max_control_connections: rs_cfg.max_control_connections as usize,
                                max_pending_external: rs_cfg.max_pending_external as usize,
                                read_timeout_ms: rs_cfg.read_timeout_ms,
                                socks5_target: None,
                                client_framing:
                                    eggress_protocol_reverse::compat_pproxy::PproxyBackwardFraming::Raw,
                            };
                        let server =
                            eggress_protocol_reverse::compat_pproxy::PproxyBackwardServer::new(
                                server_config,
                            );
                        let server_cancel = server.cancel_token();
                        let cancel_clone = cancel.clone();
                        tasks.spawn(async move {
                            let result = tokio::select! {
                                r = server.run() => r,
                                _ = cancel_clone.cancelled() => {
                                    server_cancel.cancel();
                                    Ok(())
                                }
                            };
                            if let Err(e) = result {
                                tracing::error!(error = %e, "pproxy backward server error");
                            }
                        });
                        continue;
                    }
                    let server_config = eggress_protocol_reverse::server::ReverseServerConfig {
                        control_bind: rs_cfg.control_bind,
                        external_bind: Some(rs_cfg.external_bind),
                        auth_username: rs_cfg.auth_username.clone(),
                        auth_password: rs_cfg.auth_password.clone(),
                        max_control_connections: rs_cfg.max_control_connections,
                        read_timeout_ms: rs_cfg.read_timeout_ms,
                        allow_bind: rs_cfg.allow_bind.clone(),
                        max_listeners_per_client: rs_cfg.max_listeners_per_client,
                        max_streams_per_listener: rs_cfg.max_streams_per_listener,
                        max_pending_external: rs_cfg.max_pending_external,
                    };
                    // Defense-in-depth: validate the configuration before
                    // spawning the task so unsafe configurations fail at
                    // startup rather than silently at bind time.
                    if let Err(e) = server_config.validate() {
                        tracing::error!(
                            server_id = %rs_cfg.id,
                            error = %e,
                            "reverse server configuration validation failed; skipping",
                        );
                        continue;
                    }
                    let mut server =
                        eggress_protocol_reverse::server::ReverseServer::new(server_config);
                    server.set_metrics(state_ref.reverse_metrics.clone());
                    let server_state = server.state_handle();
                    let server_cancel = server.cancel_token();

                    state_ref
                        .reverse_registry
                        .register(eggress_admin::ReverseServerEntry {
                            id: eggress_admin::ReverseServerId::from(rs_cfg.id.as_str()),
                            control_bind: rs_cfg.control_bind.to_string(),
                            state: server_state,
                        });

                    let cancel_clone = cancel.clone();
                    tasks.spawn(async move {
                        let result = tokio::select! {
                            r = server.run() => r,
                            _ = cancel_clone.cancelled() => {
                                server_cancel.cancel();
                                Ok(())
                            }
                        };
                        if let Err(e) = result {
                            tracing::error!(error = %e, "reverse server error");
                        }
                    });
                }

                for rc_cfg in reverse_clients {
                    let host = rc_cfg
                        .default_target_host
                        .clone()
                        .unwrap_or_else(|| "127.0.0.1".to_string());
                    let port = rc_cfg.default_target_port.unwrap_or(0);

                    let parallel = rc_cfg.parallel_connections.max(1);
                    for conn_idx in 0..parallel {
                        if rc_cfg.pproxy_compat {
                            let client_config = eggress_protocol_reverse::compat_pproxy::PproxyBackwardClientConfig {
                                server_addr: rc_cfg.server_addr,
                                server_chain: rc_cfg.server_chain.clone(),
                                auth: eggress_protocol_reverse::compat_pproxy::raw_auth(
                                    rc_cfg.auth_username.as_deref(),
                                    rc_cfg.auth_password.as_deref(),
                                ),
                                reconnect_initial_ms: rc_cfg.reconnect_initial_ms,
                                reconnect_max_ms: rc_cfg.reconnect_max_ms,
                                read_timeout_ms: rc_cfg.read_timeout_ms,
                                target_connect_timeout_ms: 10_000,
                                server_framing:
                                    eggress_protocol_reverse::compat_pproxy::PproxyBackwardFraming::Raw,
                            };
                            let client =
                                eggress_protocol_reverse::compat_pproxy::PproxyBackwardClient::new(
                                    client_config,
                                    std::sync::Arc::new(
                                        crate::reverse::RouteEngineTargetResolver::new(
                                            routing.clone(),
                                            host.clone(),
                                            port,
                                            std::sync::Arc::from(rc_cfg.id.as_str()),
                                            Some(rc_cfg.server_addr),
                                        ),
                                    ),
                                );
                            let cancel_clone = cancel.clone();
                            let client_cancel = client.cancel_token();
                            let client_id = rc_cfg.id.clone();
                            let server_addr = rc_cfg.server_addr;
                            tasks.spawn(async move {
                                let result = tokio::select! {
                                    r = client.run() => r,
                                    _ = cancel_clone.cancelled() => {
                                        client_cancel.cancel();
                                        Ok(())
                                    }
                                };
                                if let Err(e) = result {
                                    tracing::error!(error = %e, client_id = %client_id, server = %server_addr, conn = conn_idx, "pproxy backward client error");
                                }
                            });
                            continue;
                        }
                        let client_config = eggress_protocol_reverse::client::ReverseClientConfig {
                            server_addr: rc_cfg.server_addr,
                            auth_username: rc_cfg.auth_username.clone(),
                            auth_password: rc_cfg.auth_password.clone(),
                            reconnect_initial_ms: rc_cfg.reconnect_initial_ms,
                            reconnect_max_ms: rc_cfg.reconnect_max_ms,
                            default_target_host: rc_cfg.default_target_host.clone(),
                            default_target_port: rc_cfg.default_target_port,
                            read_timeout_ms: rc_cfg.read_timeout_ms,
                            drain_grace_ms: rc_cfg.drain_grace_ms,
                            ..Default::default()
                        };
                        let mut client =
                            eggress_protocol_reverse::client::ReverseClient::new(client_config);
                        client.set_metrics(state_ref.reverse_metrics.clone());

                        let resolver = crate::reverse::RouteEngineTargetResolver::new(
                            routing.clone(),
                            host.clone(),
                            port,
                            std::sync::Arc::from(rc_cfg.id.as_str()),
                            Some(rc_cfg.server_addr),
                        );
                        client.set_resolver(std::sync::Arc::new(resolver));

                        let cancel_clone = cancel.clone();
                        let client_cancel = client.cancel_token();
                        let client_id = rc_cfg.id.clone();
                        let server_addr = rc_cfg.server_addr;

                        tasks.spawn(async move {
                            let result = tokio::select! {
                                r = client.run() => r,
                                _ = cancel_clone.cancelled() => {
                                    client_cancel.cancel();
                                    Ok(())
                                }
                            };
                            if let Err(e) = result {
                                tracing::error!(error = %e, client_id = %client_id, server = %server_addr, conn = conn_idx, "reverse client error");
                            }
                        });
                    }
                }
            }

            // Pre-bind admin listener before marking readiness so bind failures
            // are surfaced as startup errors rather than silent background failures.
            #[cfg(feature = "operations")]
            let pre_bound_admin = if let Some(ref admin_cfg) = admin_config {
                if admin_cfg.enabled {
                    let bind = admin_cfg.bind.clone();
                    let admin_cancel_token = admin_cancel.clone();
                    match eggress_admin::AdminServer::new(&bind, admin_cancel_token).await {
                        Ok(s) => Some(s),
                        Err(e) => {
                            return Err(RuntimeError::ListenerBind {
                                addr: bind,
                                source: std::io::Error::new(
                                    std::io::ErrorKind::AddrInUse,
                                    e.to_string(),
                                ),
                            });
                        }
                    }
                } else {
                    None
                }
            } else {
                None
            };

            #[cfg(feature = "operations")]
            if let (Some(server), Some(admin_cfg)) = (pre_bound_admin, admin_config.as_ref()) {
                let metrics_enabled = admin_cfg.metrics;
                let state_ref = state_ref.clone();
                let provider: Arc<dyn AdminSnapshotProvider> = listener_infos_provider.clone();
                if let Ok(addr) = server.local_addr() {
                    match state_ref.admin_local_addr.lock() {
                        Ok(mut guard) => *guard = Some(addr),
                        Err(error) => {
                            tracing::warn!(
                                "admin listener address state was poisoned; resetting it: {error}"
                            );
                            let mut guard = error.into_inner();
                            *guard = Some(addr);
                            state_ref.admin_local_addr.clear_poison();
                        }
                    }
                }
                let admin_auth = admin_cfg.auth.clone();
                admin_tasks.spawn(async move {
                    let admin_state = eggress_admin::AdminState {
                        metrics: metrics_registry.clone(),
                        start_time: state_ref.start_time,
                        readiness: state_ref.readiness.clone(),
                        active_connections: Some(state_ref.active_connections.clone()),
                        provider,
                        udp_registry: state_ref.udp_registry.clone(),
                        #[cfg(feature = "reverse")]
                        reverse_registry: state_ref.reverse_registry.clone(),
                        #[cfg(not(feature = "reverse"))]
                        reverse_registry: std::sync::Arc::new(eggress_admin::ReverseRegistry::new()),
                        metrics_enabled,
                        auth: admin_auth,
                    };
                    if let Err(e) = server.run(admin_state).await {
                        tracing::error!("admin server error: {e}");
                    }
                });
            }

            #[cfg(unix)]
            {
                let mut sigterm =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate());
                let mut sighup =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup());

                if let Err(ref e) = sigterm {
                    tracing::warn!("failed to register SIGTERM handler: {e}");
                }
                if let Err(ref e) = sighup {
                    tracing::warn!("failed to register SIGHUP handler: {e}");
                }

                // Readiness also means signal handling is installed. This
                // prevents a reload signal from racing startup.
                readiness.store(true, Ordering::Release);

                loop {
                    tokio::select! {
                        _ = cancel.cancelled() => {
                            tracing::info!("shutdown requested via cancel token");
                            break;
                        }
                        _ = tokio::signal::ctrl_c() => {
                            tracing::info!("shutdown signal received");
                            break;
                        }
                        _ = async { sigterm.as_mut().ok()?.recv().await }, if sigterm.is_ok() => {
                            tracing::info!("shutdown signal received");
                            break;
                        }
                        _ = async { sighup.as_mut().ok()?.recv().await }, if sighup.is_ok() && !config_path.is_empty() => {
                            tracing::info!("reload signal received, reloading config from {config_path}");
                            let prev_snapshot = snapshot.load();
                            let prev_ref: Option<&CompiledRuntimeSnapshot> = Some(&prev_snapshot);
                            let config_path_clone = config_path.clone();
                            let load_result = tokio::task::spawn_blocking(move || {
                                eggress_config::compile::load_and_compile(&config_path_clone)
                            }).await;
                            match load_result {
                                Ok(Ok(new_rt_config)) => {
                                    // Classify unsupported changes before building a new snapshot.
                                    if let Err(reason) = classify_reload_config(
                                        &prev_snapshot.listeners,
                                        &prev_snapshot.timeouts,
                                        prev_snapshot.admin.as_ref(),
                                        &new_rt_config,
                                    ) {
                                        tracing::error!("reload rejected: {reason}");
                                        metrics.record_reload(false);
                                        continue;
                                    }
                                    match compile_runtime_snapshot(&new_rt_config, prev_ref) {
                                        Ok(new_snapshot) => {
                                            let upstream_count = new_snapshot.upstreams.len();
                                            let gen = new_snapshot.generation;

                                            // Snapshot must be published before the router swap so any reader
                                            // observing the new generation via
                                            // `snapshot.load()` also sees the
                                            // matching router.
                                            let new_snapshot = Arc::new(new_snapshot);
                                            snapshot.store(new_snapshot.clone());
                                            routing.swap_arc(new_snapshot.router.clone());
                                            #[cfg(feature = "operations")]
                                            state_ref.publish_admin_snapshot(new_snapshot.clone());

                                            metrics.set_config_generation(gen);
                                            metrics.record_reload(true);
                                            eggress_protocol_http::H2_POOL_REGISTRY.clear();

                                            if let Ok(mut guard) = health_clone.lock() {
                                                if let Some(ref mut hm) = *guard {
                                                    hm.stop_all();
                                                }
                                                let upstream_runtimes: Vec<Arc<UpstreamRuntime>> = snapshot
                                                    .load()
                                                    .upstreams
                                                    .values()
                                                    .cloned()
                                                    .collect();
                                                if !upstream_runtimes.is_empty() {
                                                    let mut hm = HealthManager::new(health_cancel.clone());
                                                    hm.start_probes(&upstream_runtimes);
                                                    *guard = Some(hm);
                                                } else {
                                                    *guard = None;
                                                }
                                            }

                                            tracing::info!(
                                                generation = gen,
                                                upstreams = upstream_count,
                                                "config reloaded successfully"
                                            );
                                        }
                                        Err(e) => {
                                            metrics.record_reload(false);
                                            tracing::error!("reload failed (snapshot build): {e}");
                                        }
                                    }
                                }
                                Ok(Err(e)) => {
                                    metrics.record_reload(false);
                                    tracing::error!("reload failed (config load): {e}");
                                }
                                Err(join_err) => {
                                    metrics.record_reload(false);
                                    tracing::error!("reload task panicked: {join_err}");
                                }
                            }
                        }
                    }
                }
            }

            #[cfg(not(unix))]
            {
                readiness.store(true, Ordering::Release);
                tokio::select! {
                    _ = cancel.cancelled() => {
                        tracing::info!("shutdown requested via cancel token");
                    }
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("shutdown signal received");
                    }
                }
            }

            // 1. Set readiness false (admin /-/ready will report 503 during drain)
            readiness.store(false, Ordering::Release);

            // 2. Stop listeners (no new connections accepted)
            listener_cancel.cancel();

            // 3. Stop health probes
            health_cancel.cancel();

            // 4. Close all UDP associations
            state_ref.udp_registry.close_all().await;

            // 5. Wait for UDP relay tasks to complete
            state_ref.udp_tasks.close();
            let _ = tokio::time::timeout(shutdown_grace, state_ref.udp_tasks.wait()).await;

            // 6. Wait for listener accept loops to exit so they cannot hand
            //    new connections to the connection tracker.
            tasks.close();
            tasks.wait().await;

            // 7. Drain active connections within the grace period; force-cancel
            //    afterwards. Admin stays up through this window so operators
            //    can observe drain progress via /-/ready, /-/status, /metrics.
            tracing::info!("draining active connections");

            let deadline = tokio::time::Instant::now() + shutdown_grace;
            loop {
                let active = active_connections.load(Ordering::Acquire);
                if active == 0 {
                    tracing::info!("all connections drained");
                    break;
                }
                if tokio::time::Instant::now() >= deadline {
                    tracing::warn!(active, "drain timeout reached, forcing shutdown");
                    connection_cancel.cancel();
                    break;
                }
                tokio::time::sleep(Duration::from_millis(100)).await;
            }

            // 8. Wait for connection tasks (either drained naturally or force-cancelled)
            connection_tasks.close();
            connection_tasks.wait().await;

            #[cfg(feature = "ssh")]
            ssh_sessions.shutdown().await;

            // 9. Now that the proxy has fully stopped accepting and serving
            //    traffic, stop the admin server. /-/ready has been reporting
            //    503 since step 1.
            admin_cancel.cancel();
            admin_tasks.close();
            admin_tasks.wait().await;

            #[cfg(feature = "operations")]
            if let Some(mut proxy) = compatibility_system_proxy {
                proxy.restore().map_err(RuntimeError::Other)?;
            }

            Ok::<_, RuntimeError>(())
        };

        let result = if tokio::runtime::Handle::try_current().is_err() {
            // Caller is not inside a tokio runtime; create one and block on it.
            let rt = tokio::runtime::Runtime::new()?;
            rt.block_on(run_async)
        } else {
            // Caller is already inside a tokio runtime. Driving a long-lived
            // supervisor body via Handle::current().block_on from a worker
            // thread would panic or deadlock. Run on a dedicated OS thread
            // with its own runtime.
            std::thread::Builder::new()
                .name("eggress-supervisor".to_string())
                .spawn(move || -> Result<(), RuntimeError> {
                    let rt = tokio::runtime::Builder::new_multi_thread()
                        .enable_all()
                        .build()
                        .map_err(RuntimeError::RuntimeInit)?;
                    rt.block_on(run_async)
                })
                .map_err(RuntimeError::RuntimeInit)?
                .join()
                .map_err(|payload| {
                    let message = payload
                        .downcast_ref::<String>()
                        .map(String::as_str)
                        .or_else(|| payload.downcast_ref::<&'static str>().copied())
                        .unwrap_or("unknown panic payload");
                    RuntimeError::Other(format!("supervisor thread panicked: {message}"))
                })?
        };

        match &result {
            Ok(()) => tracing::info!("eggress stopped"),
            Err(e) => tracing::error!(error = %e, "eggress stopped with error"),
        }

        result
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use tempfile::NamedTempFile;

    use crate::snapshot::compile_runtime_snapshot;
    use eggress_config::compile::{GroupFallback, ProcessConfig, RuntimeConfig, TimeoutConfig};
    use eggress_routing::scheduler::SchedulerKind;
    use eggress_routing::{MatchExpr, RouteActionSpec, RuleId, UpstreamGroupId};

    fn write_config(content: &str) -> NamedTempFile {
        let mut f = NamedTempFile::new().unwrap();
        f.write_all(content.as_bytes()).unwrap();
        f.flush().unwrap();
        f
    }

    #[test]
    fn build_router_direct_only() {
        let rt_config = RuntimeConfig {
            process: ProcessConfig::default(),
            timeouts: TimeoutConfig::default(),
            listeners: vec![],
            upstreams: vec![],
            groups: vec![],
            rules: vec![],
            default_action: RouteActionSpec::Direct,
            admin: None,
            reverse_servers: vec![],
            reverse_clients: vec![],
        };
        let snap = compile_runtime_snapshot(&rt_config, None).unwrap();
        assert!(snap.router.rules().is_empty());
    }

    #[test]
    fn build_router_with_group_references_unknown_upstream() {
        let rt_config = RuntimeConfig {
            process: ProcessConfig::default(),
            timeouts: TimeoutConfig::default(),
            listeners: vec![],
            upstreams: vec![],
            groups: vec![eggress_config::compile::UpstreamGroupConfig {
                id: UpstreamGroupId(Arc::from("main")),
                scheduler: SchedulerKind::RoundRobin,
                members: vec!["nonexistent".to_string()],
                fallback: GroupFallback::Reject,
            }],
            rules: vec![],
            default_action: RouteActionSpec::Direct,
            admin: None,
            reverse_servers: vec![],
            reverse_clients: vec![],
        };
        let result = compile_runtime_snapshot(&rt_config, None);
        assert!(result.is_err(), "expected error, got Ok");
        let err_msg = result.err().unwrap().to_string();
        assert!(err_msg.contains("nonexistent"));
    }

    #[test]
    fn build_router_with_valid_group() {
        let rt_config = RuntimeConfig {
            process: ProcessConfig::default(),
            timeouts: TimeoutConfig::default(),
            listeners: vec![],
            upstreams: vec![eggress_config::compile::UpstreamConfig {
                id: "proxy1".to_string(),
                chain: eggress_uri::ProxyChainSpec { hops: vec![] },
                health: eggress_routing::health::HealthConfig::default(),
                h2: None,
            }],
            groups: vec![eggress_config::compile::UpstreamGroupConfig {
                id: UpstreamGroupId(Arc::from("main")),
                scheduler: SchedulerKind::RoundRobin,
                members: vec!["proxy1".to_string()],
                fallback: GroupFallback::Reject,
            }],
            rules: vec![],
            default_action: RouteActionSpec::Direct,
            admin: None,
            reverse_servers: vec![],
            reverse_clients: vec![],
        };
        let snap = compile_runtime_snapshot(&rt_config, None).unwrap();
        assert!(snap.router.rules().is_empty());
    }

    #[test]
    fn build_router_rule_references_unknown_group() {
        let rt_config = RuntimeConfig {
            process: ProcessConfig::default(),
            timeouts: TimeoutConfig::default(),
            listeners: vec![],
            upstreams: vec![],
            groups: vec![],
            rules: vec![eggress_routing::CompiledRule {
                id: RuleId(Arc::from("r1")),
                matcher: MatchExpr::Any,
                action: RouteActionSpec::UpstreamGroup(UpstreamGroupId(Arc::from("missing"))),
            }],
            default_action: RouteActionSpec::Direct,
            admin: None,
            reverse_servers: vec![],
            reverse_clients: vec![],
        };
        let result = compile_runtime_snapshot(&rt_config, None);
        assert!(result.is_err(), "expected error, got Ok");
        let err_msg = result.err().unwrap().to_string();
        assert!(err_msg.contains("missing"));
    }

    #[tokio::test]
    async fn load_config_start_supervisor() {
        let config = r#"
version = 1

[[listeners]]
name = "test"
bind = "127.0.0.1:0"
protocols = ["http"]
"#;
        let f = write_config(config);
        let path = f.path().to_str().unwrap();
        let result = ServiceSupervisor::start(path);
        assert!(result.is_ok(), "expected Ok, got {:?}", result.err());
    }

    #[test]
    fn active_connections_counter_increments_and_decrements() {
        let active = Arc::new(AtomicU64::new(0));
        assert_eq!(active.load(Ordering::Acquire), 0);
        active.fetch_add(1, Ordering::AcqRel);
        assert_eq!(active.load(Ordering::Acquire), 1);
        active.fetch_add(1, Ordering::AcqRel);
        assert_eq!(active.load(Ordering::Acquire), 2);
        active.fetch_sub(1, Ordering::Release);
        assert_eq!(active.load(Ordering::Acquire), 1);
        active.fetch_sub(1, Ordering::Release);
        assert_eq!(active.load(Ordering::Acquire), 0);
    }

    #[test]
    fn active_connections_guard_releases_on_panic() {
        let active = Arc::new(AtomicU64::new(0));
        let result = std::panic::catch_unwind({
            let active = active.clone();
            move || {
                let _guard = ActiveConnectionGuard::new(active);
                panic!("connection task panic");
            }
        });
        assert!(result.is_err());
        assert_eq!(active.load(Ordering::Acquire), 0);
    }

    #[test]
    fn readiness_flag_controls_ready_endpoint() {
        let readiness = Arc::new(AtomicBool::new(true));
        assert!(readiness.load(Ordering::Relaxed));
        readiness.store(false, Ordering::Relaxed);
        assert!(!readiness.load(Ordering::Relaxed));
        readiness.store(true, Ordering::Relaxed);
        assert!(readiness.load(Ordering::Relaxed));
    }

    #[test]
    fn reload_rejects_listener_name_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-changed"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let path1 = f1.path().to_str().unwrap();
        let path2 = f2.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path1).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(result.is_err(), "listener name change should be rejected");
        assert!(result.unwrap_err().contains("name changed"));
    }

    #[test]
    fn reload_rejects_listener_bind_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:9090"
protocols = ["http"]
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let path1 = f1.path().to_str().unwrap();
        let path2 = f2.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path1).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(result.is_err(), "listener bind change should be rejected");
        assert!(result.unwrap_err().contains("bind"));
    }

    #[test]
    fn reload_accepts_unchanged_listeners() {
        let config = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let f = write_config(config);
        let path = f.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(result.is_ok(), "unchanged listeners should be accepted");
    }

    #[test]
    fn reload_rejects_udp_topology_changes() {
        let config_without_udp = r#"
version = 1

[[listeners]]
name = "socks-in"
bind = "127.0.0.1:8080"
protocols = ["socks5"]
"#;
        let config_with_udp = r#"
version = 1

[[listeners]]
name = "socks-in"
bind = "127.0.0.1:8080"
protocols = ["socks5"]

[listeners.udp]
enabled = true
bind = "127.0.0.1:0"
"#;
        let f1 = write_config(config_without_udp);
        let f2 = write_config(config_with_udp);
        let sup = ServiceSupervisor::start(f1.path().to_str().unwrap()).unwrap();
        let new_config =
            eggress_config::compile::load_and_compile(f2.path().to_str().unwrap()).unwrap();

        let result = sup.classify_reload(&new_config);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("UDP"));
    }

    #[test]
    fn reload_rejects_timeout_change() {
        let config1 = r#"
version = 1

[timeouts]
handshake = "10s"

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let config2 = r#"
version = 1

[timeouts]
handshake = "5s"

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let sup = ServiceSupervisor::start(f1.path().to_str().unwrap()).unwrap();
        let new_config =
            eggress_config::compile::load_and_compile(f2.path().to_str().unwrap()).unwrap();

        let result = sup.classify_reload(&new_config);
        assert!(result.is_err());
        assert!(result.unwrap_err().contains("timeout"));
    }

    #[test]
    fn reload_rejects_admin_bind_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[admin]
bind = "127.0.0.1:9090"
enabled = false
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[admin]
bind = "127.0.0.1:9091"
enabled = false
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let sup = ServiceSupervisor::start(f1.path().to_str().unwrap()).unwrap();
        let new_config =
            eggress_config::compile::load_and_compile(f2.path().to_str().unwrap()).unwrap();

        let result = sup.classify_reload(&new_config);
        assert!(result.is_err(), "admin bind change should be rejected");
        assert!(result.unwrap_err().contains("admin"));
    }

    #[test]
    fn reload_rejects_listener_count_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[[listeners]]
name = "socks-in"
bind = "127.0.0.1:1080"
protocols = ["socks5"]
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let path1 = f1.path().to_str().unwrap();
        let path2 = f2.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path1).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(result.is_err(), "listener count change should be rejected");
        assert!(result.unwrap_err().contains("listener count"));
    }

    #[test]
    fn reload_rejects_transparent_enabled_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[listeners.transparent]
enabled = true
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let path1 = f1.path().to_str().unwrap();
        let path2 = f2.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path1).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(
            result.is_err(),
            "transparent enabled change should be rejected"
        );
        assert!(result.unwrap_err().contains("transparent"));
    }

    #[test]
    fn reload_rejects_unix_path_change() {
        let config1 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[listeners.unix]
path = "/tmp/eggress.sock"
"#;
        let config2 = r#"
version = 1

[[listeners]]
name = "http-in"
bind = "127.0.0.1:8080"
protocols = ["http"]

[listeners.unix]
path = "/tmp/eggress-new.sock"
"#;
        let f1 = write_config(config1);
        let f2 = write_config(config2);
        let path1 = f1.path().to_str().unwrap();
        let path2 = f2.path().to_str().unwrap();

        let sup = ServiceSupervisor::start(path1).unwrap();
        let new_config = eggress_config::compile::load_and_compile(path2).unwrap();
        let result = sup.classify_reload(&new_config);
        assert!(result.is_err(), "unix path change should be rejected");
        assert!(result.unwrap_err().contains("unix socket path"));
    }

    #[test]
    fn compute_advertise_explicit() {
        let result = compute_advertise_ip(
            Some("10.0.0.1".parse().unwrap()),
            "0.0.0.0".parse().unwrap(),
            Some("127.0.0.1:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V4("10.0.0.1".parse().unwrap())
        );
    }

    #[test]
    fn compute_advertise_bind_ip() {
        let result = compute_advertise_ip(
            None,
            "192.168.1.1".parse().unwrap(),
            Some("127.0.0.1:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V4("192.168.1.1".parse().unwrap())
        );
    }

    #[test]
    fn compute_advertise_loopback_fallback() {
        let result = compute_advertise_ip(
            None,
            "0.0.0.0".parse().unwrap(),
            Some("127.0.0.1:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
        );
    }

    #[test]
    fn compute_advertise_unspecified_non_loopback_rejected() {
        let result = compute_advertise_ip(
            None,
            "0.0.0.0".parse().unwrap(),
            Some("192.168.1.10:5000".parse().unwrap()),
        );
        assert!(
            result.is_err(),
            "non-loopback with unspecified bind should fail"
        );
    }

    #[test]
    fn compute_advertise_without_tcp_peer_is_rejected() {
        let result = compute_advertise_ip(None, "0.0.0.0".parse().unwrap(), None);
        assert!(result.is_err());
    }

    #[test]
    fn compute_advertise_ipv6_loopback() {
        let result = compute_advertise_ip(
            None,
            "::".parse().unwrap(),
            Some("[::1]:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V6(std::net::Ipv6Addr::LOCALHOST)
        );
    }

    #[test]
    fn compute_advertise_preserves_unspecified_bind_family() {
        let result = compute_advertise_ip(
            None,
            "0.0.0.0".parse().unwrap(),
            Some("[::ffff:127.0.0.1]:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)
        );
    }

    #[test]
    fn compute_advertise_explicit_overrides_bind() {
        let result = compute_advertise_ip(
            Some("10.0.0.1".parse().unwrap()),
            "192.168.1.1".parse().unwrap(),
            Some("127.0.0.1:5000".parse().unwrap()),
        );
        assert_eq!(
            result.unwrap(),
            std::net::IpAddr::V4("10.0.0.1".parse().unwrap())
        );
    }
}