rama-net 0.4.0

//! Route client requests through the operating system's proxy settings.

use std::{
    fmt,
    sync::{
        Arc, OnceLock,
        atomic::{AtomicU64, Ordering},
    },
    time::{Duration, Instant},
};

use arc_swap::ArcSwapOption;
use rama_core::{
    Layer, Service,
    error::{BoxError, BoxErrorExt as _, ErrorContext},
    error_sink::{ErrorSink, TracingErrorSink},
    extensions::{Extensions, ExtensionsRef},
    service::{BoxService, service_fn},
};
use rama_utils::macros::generate_set_and_with;

#[cfg(any(
    test,
    target_vendor = "apple",
    target_os = "android",
    target_os = "linux",
    target_os = "freebsd",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "dragonfly"
))]
use crate::address::{Host, HostWithPort};
use crate::{
    Protocol,
    address::{Authority, HostRef, HostWithOptPort, ProxyAddress},
    input_ext::{AuthorityInputExt, ProtocolInputExt, UriInputExt},
    uri::Uri,
};

use super::{
    ProxyRoute, ProxyRoutes,
    bypass::{BypassRule, BypassRuleDialect, is_simple_hostname, matches_any_rule},
    load::LoadErrorPolicy,
};

mod platform;

/// How long a [`SystemProxyLayer`] keeps a system proxy snapshot before lazily
/// checking for changes.
///
/// The ten-second default follows the polling interval used by
/// [Chromium's Windows proxy configuration service][chromium] where change
/// notifications alone are insufficient. Use
/// [`SystemProxyLayer::new_with_ttl`] to select a different value.
///
/// [chromium]: https://chromium.googlesource.com/chromium/src/+/refs/heads/main/net/proxy_resolution/win/proxy_config_service_win.cc
pub const DEFAULT_SYSTEM_PROXY_CONFIG_TTL: Duration = Duration::from_secs(10);

/// How system proxy discovery handles bypass rules Rama cannot parse.
///
/// This policy applies independently of whether a platform uses ordinary or
/// reversed bypass-list semantics. Rejecting invalid rules prevents a partial
/// snapshot from making routing decisions with a different rule set than the
/// operating system supplied.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum SystemProxyInvalidBypassRulePolicy {
    /// Ignore an invalid rule and retain the rest of the system snapshot.
    #[default]
    Ignore,
    /// Reject the complete system snapshot when any bypass rule is invalid.
    Reject,
}

/// The request information passed to a system PAC resolver.
///
/// When produced by [`SystemProxyLayer`], the URI is absolute, has a root path
/// when the original target omitted one, and omits the scheme's default port.
/// The extension store is a cheap clone of the input's store. This keeps caller
/// metadata available to custom PAC implementations without borrowing the
/// request across an await point.
#[derive(Debug, Clone)]
pub struct SystemProxyPacRequest {
    /// Metadata cloned from the routed service input.
    pub extensions: Extensions,
    /// The normalized absolute URI for which routes are requested.
    pub uri: Uri,
}

impl SystemProxyPacRequest {
    /// Create a PAC request from an absolute request URI and its extensions.
    pub fn new(extensions: Extensions, uri: Uri) -> Result<Self, BoxError> {
        if !uri.is_absolute() || uri.host().is_none() {
            return Err(BoxError::from_static_str(
                "system proxy PAC request URI must be absolute and have a host",
            ));
        }
        Ok(Self { extensions, uri })
    }
}

impl ExtensionsRef for SystemProxyPacRequest {
    fn extensions(&self) -> &Extensions {
        &self.extensions
    }
}

impl UriInputExt for SystemProxyPacRequest {
    fn uri(&self) -> &Uri {
        &self.uri
    }
}

/// Resolves proxy routes for a request using one system-configured PAC script.
///
/// Returning `None` asks the system layer to try the fixed proxy settings from
/// the same snapshot, if any, and otherwise leave the request unchanged.
/// The blanket implementation accepts any resolver error that converts into
/// [`BoxError`]. Implementations may return a concrete service; no allocation
/// or type erasure is required.
pub trait SystemProxyPacResolver:
    Service<SystemProxyPacRequest, Output = Option<ProxyRoutes>, Error: Into<BoxError>>
{
}

impl<T> SystemProxyPacResolver for T where
    T: Service<SystemProxyPacRequest, Output = Option<ProxyRoutes>, Error: Into<BoxError>>
{
}

/// Supplies a resolver for a system-configured PAC URI.
///
/// The blanket implementation accepts any factory and resolver errors that
/// convert into [`BoxError`]. The resolver output remains concrete so
/// implementations can choose their own caching and sharing strategy.
pub trait SystemProxyPacService:
    Service<Uri, Error: Into<BoxError>, Output: SystemProxyPacResolver>
{
}

impl<T> SystemProxyPacService for T where
    T: Service<Uri, Error: Into<BoxError>, Output: SystemProxyPacResolver>
{
}

/// A snapshot of the operating system's proxy configuration.
///
/// HTTP and HTTPS identify the destination scheme, not necessarily the
/// transport protocol used to reach the proxy. A SOCKS5 proxy is used as a
/// fallback when no scheme-specific proxy is configured. A PAC URI takes
/// precedence over fixed proxies because it can make a per-request decision;
/// each platform reader records whether its bypass entries apply before PAC.
/// System proxy routing always bypasses loopback hosts, including before PAC
/// evaluation, matching native proxy stacks.
#[derive(Debug, Clone, Default)]
pub struct SystemProxyConfig {
    http: Option<ProxyAddress>,
    https: Option<ProxyAddress>,
    socks5: Option<ProxyAddress>,
    pac_uri: Option<Uri>,
    auto_detect: bool,
    bypass: Arc<[BypassRule]>,
    exclude_simple_hostnames: bool,
    reversed_bypass: bool,
    bypass_before_pac: bool,
}

impl SystemProxyConfig {
    fn replace_bypass_ignoring_invalid(
        &mut self,
        bypass: impl IntoIterator<Item = impl Into<Box<str>>>,
        dialect: BypassRuleDialect,
    ) {
        self.bypass = bypass
            .into_iter()
            .filter_map(
                |value| match BypassRule::compile_with_dialect(value, dialect) {
                    Ok(rule) => Some(rule),
                    Err(error) => {
                        rama_core::telemetry::tracing::debug!(
                            error = %error,
                            "ignoring invalid system proxy bypass pattern"
                        );
                        None
                    }
                },
            )
            .collect();
    }

    fn try_replace_bypass(
        &mut self,
        bypass: impl IntoIterator<Item = impl Into<Box<str>>>,
        policy: SystemProxyInvalidBypassRulePolicy,
        dialect: BypassRuleDialect,
    ) -> Result<(), BoxError> {
        match policy {
            SystemProxyInvalidBypassRulePolicy::Ignore => {
                self.replace_bypass_ignoring_invalid(bypass, dialect);
            }
            SystemProxyInvalidBypassRulePolicy::Reject => {
                self.bypass = bypass
                    .into_iter()
                    .map(|value| BypassRule::compile_with_dialect(value, dialect))
                    .collect::<Result<Vec<_>, _>>()?
                    .into();
            }
        }
        Ok(())
    }

    #[cfg(any(
        test,
        target_vendor = "apple",
        target_os = "android",
        target_os = "windows",
        target_os = "linux",
        target_os = "freebsd",
        target_os = "netbsd",
        target_os = "openbsd",
        target_os = "dragonfly"
    ))]
    fn try_set_bypass_with_dialect(
        &mut self,
        bypass: impl IntoIterator<Item = impl Into<Box<str>>>,
        policy: SystemProxyInvalidBypassRulePolicy,
        dialect: BypassRuleDialect,
    ) -> Result<(), BoxError> {
        self.try_replace_bypass(bypass, policy, dialect)
    }

    /// Read the current platform proxy snapshot.
    ///
    /// - Windows uses the active user's WinINET/Internet Options settings;
    /// - macOS and iOS use CFNetwork's system proxy dictionary;
    /// - Android uses `ConnectivityManager.getDefaultProxy()`, with the legacy
    ///   `Proxy` API on Android versions before API 23;
    /// - Linux and BSD prefer KDE's `kioslaverc` on KDE desktops, otherwise
    ///   reading GNOME `gsettings` before falling back to KDE.
    ///
    /// This deliberately does not inspect `http_proxy` or related environment
    /// variables. Those are application configuration and are handled by
    /// [`ProxyEnvLayer`][crate::client::ProxyEnvLayer] and
    /// [`NoProxyEnvLayer`][crate::client::NoProxyEnvLayer] instead.
    ///
    /// Automatic discovery such as WPAD is recorded by [`Self::auto_detect`]
    /// but is not attempted when the platform does not provide a concrete PAC
    /// URI. Malformed non-empty proxy values are reported as errors rather
    /// than silently bypassing a configured system policy.
    ///
    /// Platform operations that have asynchronous APIs are awaited directly.
    /// Native platforms that only expose a synchronous snapshot call keep that
    /// call narrowly isolated inside their platform reader.
    pub async fn try_from_system() -> Result<Self, BoxError> {
        Self::try_from_system_with_invalid_bypass_rule_policy(
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
        .await
    }

    /// Read the current platform proxy snapshot with an explicit invalid
    /// bypass-rule policy.
    ///
    /// [`Ignore`][SystemProxyInvalidBypassRulePolicy::Ignore] is the default
    /// used by [`Self::try_from_system`]. Selecting
    /// [`Reject`][SystemProxyInvalidBypassRulePolicy::Reject] returns an error
    /// instead of accepting a snapshot with one or more discarded rules.
    pub async fn try_from_system_with_invalid_bypass_rule_policy(
        policy: SystemProxyInvalidBypassRulePolicy,
    ) -> Result<Self, BoxError> {
        platform::read(policy)
            .await
            .context("read system proxy configuration")
    }

    /// Return whether this snapshot contains no automatic, PAC, or fixed proxy
    /// settings.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.http.is_none()
            && self.https.is_none()
            && self.socks5.is_none()
            && self.pac_uri.is_none()
            && !self.auto_detect
    }

    /// The proxy for HTTP destinations.
    #[must_use]
    pub const fn http_proxy(&self) -> Option<&ProxyAddress> {
        self.http.as_ref()
    }

    /// The proxy for HTTPS destinations.
    #[must_use]
    pub const fn https_proxy(&self) -> Option<&ProxyAddress> {
        self.https.as_ref()
    }

    /// The SOCKS5 fallback proxy.
    #[must_use]
    pub const fn socks5_proxy(&self) -> Option<&ProxyAddress> {
        self.socks5.as_ref()
    }

    /// The configured PAC script URI.
    #[must_use]
    pub const fn pac_uri(&self) -> Option<&Uri> {
        self.pac_uri.as_ref()
    }

    /// Whether the platform requested automatic proxy discovery (for example,
    /// WPAD) without necessarily supplying a concrete PAC URI.
    #[must_use]
    pub const fn auto_detect(&self) -> bool {
        self.auto_detect
    }

    /// Host patterns that bypass fixed proxies.
    pub fn bypass(&self) -> impl Iterator<Item = &str> {
        self.bypass.iter().map(BypassRule::raw)
    }

    /// Whether names without a dot bypass fixed proxies.
    #[must_use]
    pub const fn exclude_simple_hostnames(&self) -> bool {
        self.exclude_simple_hostnames
    }

    /// Whether fixed proxies are used only for hosts matching [`bypass`][Self::bypass].
    ///
    /// KDE exposes this uncommon inverted exception-list mode. The default is
    /// `false`, where matching hosts bypass the proxy in the usual way.
    #[must_use]
    pub const fn reversed_bypass(&self) -> bool {
        self.reversed_bypass
    }

    generate_set_and_with! {
        /// Set the proxy used for HTTP destinations.
        pub fn http_proxy(mut self, proxy: Option<ProxyAddress>) -> Self {
            self.http = proxy;
            self
        }
    }

    generate_set_and_with! {
        /// Set the proxy used for HTTPS destinations.
        pub fn https_proxy(mut self, proxy: Option<ProxyAddress>) -> Self {
            self.https = proxy;
            self
        }
    }

    generate_set_and_with! {
        /// Set the SOCKS5 fallback proxy.
        pub fn socks5_proxy(mut self, proxy: Option<ProxyAddress>) -> Self {
            self.socks5 = proxy;
            self
        }
    }

    generate_set_and_with! {
        /// Set the PAC script URI.
        pub fn pac_uri(mut self, pac_uri: Option<Uri>) -> Self {
            self.pac_uri = pac_uri;
            self
        }
    }

    generate_set_and_with! {
        /// Record whether automatic proxy discovery is enabled.
        ///
        /// Rama exposes this signal but does not perform WPAD itself.
        pub fn auto_detect(mut self, auto_detect: bool) -> Self {
            self.auto_detect = auto_detect;
            self
        }
    }

    generate_set_and_with! {
        /// Replace the fixed-proxy bypass patterns, ignoring invalid entries.
        ///
        /// Use [`Self::try_set_bypass`] with
        /// [`Reject`][SystemProxyInvalidBypassRulePolicy::Reject] when an invalid
        /// entry should reject the complete update. Boxed strings transfer
        /// directly into snapshot storage; borrowed strings are copied because
        /// the snapshot owns its rules.
        pub fn bypass(
            mut self,
            bypass: impl IntoIterator<Item = impl Into<Box<str>>>,
        ) -> Self {
            self.replace_bypass_ignoring_invalid(bypass, BypassRuleDialect::Rama);
            self
        }
    }

    generate_set_and_with! {
        /// Replace the fixed-proxy bypass patterns using an explicit invalid-rule
        /// policy.
        ///
        /// The update is atomic: when [`Reject`][SystemProxyInvalidBypassRulePolicy::Reject]
        /// encounters an invalid rule, this returns an error and leaves the prior
        /// bypass list unchanged. Boxed strings transfer directly into snapshot
        /// storage; borrowed strings are copied because the snapshot owns its
        /// rules.
        pub fn bypass(
            mut self,
            bypass: impl IntoIterator<Item = impl Into<Box<str>>>,
            policy: SystemProxyInvalidBypassRulePolicy,
        ) -> Result<Self, BoxError> {
            self.try_replace_bypass(bypass, policy, BypassRuleDialect::Rama)?;
            Ok(self)
        }
    }

    generate_set_and_with! {
        /// Configure whether names without a dot bypass fixed proxies.
        pub fn exclude_simple_hostnames(mut self, exclude: bool) -> Self {
            self.exclude_simple_hostnames = exclude;
            self
        }
    }

    generate_set_and_with! {
        /// Invert the meaning of fixed-proxy bypass patterns.
        pub fn reversed_bypass(mut self, reversed: bool) -> Self {
            self.reversed_bypass = reversed;
            self
        }
    }

    fn decision(&self, uri: &Uri) -> SystemProxyDecision {
        if let Some(pac_uri) = &self.pac_uri {
            if uri.host().is_some_and(|host| {
                host.is_loopback()
                    || (self.bypass_before_pac
                        && self.bypasses(
                            uri.scheme(),
                            host,
                            uri.port_u16()
                                .or_else(|| uri.scheme().and_then(Protocol::default_port)),
                        ))
            }) {
                return SystemProxyDecision::Route(ProxyRoute::Direct);
            }
            return SystemProxyDecision::Pac(pac_uri.clone());
        }

        self.fixed_route(uri)
            .map(SystemProxyDecision::Route)
            .unwrap_or(SystemProxyDecision::None)
    }

    fn fixed_route(&self, uri: &Uri) -> Option<ProxyRoute> {
        let host = uri.host()?;
        self.fixed_route_for(
            uri.scheme(),
            host,
            uri.port_u16()
                .or_else(|| uri.scheme().and_then(Protocol::default_port)),
        )
    }

    fn fixed_route_for(
        &self,
        scheme: Option<&Protocol>,
        host: HostRef<'_>,
        port: Option<u16>,
    ) -> Option<ProxyRoute> {
        let proxy = match scheme {
            Some(protocol) if *protocol == Protocol::HTTPS || *protocol == Protocol::WSS => {
                self.https.as_ref()
            }
            Some(protocol) if *protocol == Protocol::HTTP || *protocol == Protocol::WS => {
                self.http.as_ref()
            }
            _ => None,
        }
        .or(self.socks5.as_ref());
        let proxy = proxy?;

        // Platform proxy stacks implicitly keep loopback traffic local even
        // when their user-visible bypass list does not mention it.
        if host.is_loopback() || self.bypasses(scheme, host, port) {
            return Some(ProxyRoute::Direct);
        }
        Some(ProxyRoute::Proxy(proxy.clone()))
    }

    fn bypasses(&self, scheme: Option<&Protocol>, host: HostRef<'_>, port: Option<u16>) -> bool {
        let matches = (self.exclude_simple_hostnames && is_simple_hostname(host))
            || matches_any_rule(&self.bypass, scheme, host, port);
        if self.reversed_bypass {
            !matches
        } else {
            matches
        }
    }
}

enum SystemProxyDecision {
    None,
    Route(ProxyRoute),
    Pac(Uri),
}

type SystemProxyConfigReader = BoxService<(), SystemProxyConfig, BoxError>;
type BoxSystemProxyConfigChangeTrigger = BoxService<(), bool, BoxError>;

#[derive(Clone, Default)]
struct LazyPlatformConfigChangeTrigger {
    trigger: Arc<OnceLock<Arc<platform::PlatformConfigChangeTrigger>>>,
}

impl LazyPlatformConfigChangeTrigger {
    fn poll(&self) -> Result<bool, BoxError> {
        self.trigger
            .get_or_init(platform::config_change_trigger)
            .poll()
    }

    #[cfg(test)]
    fn is_initialized(&self) -> bool {
        self.trigger.get().is_some()
    }
}

#[derive(Clone)]
enum SystemProxyConfigChangeTrigger {
    Platform(LazyPlatformConfigChangeTrigger),
    Custom(BoxSystemProxyConfigChangeTrigger),
}

impl fmt::Debug for SystemProxyConfigChangeTrigger {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Platform(_) => f.write_str("Platform(_)"),
            Self::Custom(trigger) => f.debug_tuple("Custom").field(trigger).finish(),
        }
    }
}

#[derive(Debug, Default)]
struct RefreshRequestState {
    requested_generation: AtomicU64,
    completed_generation: AtomicU64,
}

impl RefreshRequestState {
    fn request(&self) -> u64 {
        self.requested_generation
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| {
                Some(generation.saturating_add(1))
            })
            .unwrap_or_else(|generation| generation)
            .saturating_add(1)
    }
}

#[derive(Clone)]
struct RefreshRequest {
    state: Arc<RefreshRequestState>,
    generation: u64,
}

impl RefreshRequest {
    fn is_pending(&self) -> bool {
        self.generation > self.state.completed_generation.load(Ordering::Acquire)
    }

    fn complete(&self) {
        self.state
            .completed_generation
            .fetch_max(self.generation, Ordering::AcqRel);
    }
}

#[derive(Clone)]
struct SystemProxyConfigRefresh {
    enabled: bool,
    trigger: Option<SystemProxyConfigChangeTrigger>,
    trigger_error_sink: Arc<dyn ErrorSink>,
    requests: Arc<RefreshRequestState>,
}

impl fmt::Debug for SystemProxyConfigRefresh {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SystemProxyConfigRefresh")
            .field("enabled", &self.enabled)
            .field("trigger", &self.trigger)
            .finish_non_exhaustive()
    }
}

impl Default for SystemProxyConfigRefresh {
    fn default() -> Self {
        Self {
            enabled: true,
            trigger: Some(SystemProxyConfigChangeTrigger::Platform(
                LazyPlatformConfigChangeTrigger::default(),
            )),
            trigger_error_sink: Arc::new(TracingErrorSink::default()),
            requests: Arc::new(RefreshRequestState::default()),
        }
    }
}

impl SystemProxyConfigRefresh {
    async fn requested(&self) -> RefreshRequest {
        let changed = if !self.enabled {
            false
        } else if let Some(trigger) = &self.trigger {
            match trigger {
                SystemProxyConfigChangeTrigger::Platform(trigger) => trigger.poll(),
                SystemProxyConfigChangeTrigger::Custom(trigger) => trigger.serve(()).await,
            }
            .unwrap_or_else(|error| {
                self.trigger_error_sink.sink_error(error);
                false
            })
        } else {
            false
        };
        let generation = if changed {
            self.requests.request()
        } else {
            self.requests.requested_generation.load(Ordering::Acquire)
        };
        RefreshRequest {
            state: self.requests.clone(),
            generation,
        }
    }
}

#[derive(Debug)]
struct IntoBoxErrorService<T>(T);

impl<T> Service<()> for IntoBoxErrorService<T>
where
    T: Service<(), Output = bool>,
    T::Error: Into<BoxError>,
{
    type Output = bool;
    type Error = BoxError;

    async fn serve(&self, (): ()) -> Result<Self::Output, Self::Error> {
        self.0.serve(()).await.map_err(Into::into)
    }
}

fn system_proxy_config_reader(
    policy: SystemProxyInvalidBypassRulePolicy,
) -> SystemProxyConfigReader {
    BoxService::new(service_fn(move |()| async move {
        SystemProxyConfig::try_from_system_with_invalid_bypass_rule_policy(policy).await
    }))
}

struct SystemProxyConfigCache {
    current: ArcSwapOption<SystemProxyConfig>,
    ttl: Duration,
    epoch: Instant,
    refresh_after_nanos: AtomicU64,
    cold_failure_generation: AtomicU64,
    refresh_lock: tokio::sync::Mutex<()>,
    reader: SystemProxyConfigReader,
}

impl fmt::Debug for SystemProxyConfigCache {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SystemProxyConfigCache")
            .field("current", &self.current.load_full())
            .field("ttl", &self.ttl)
            .field(
                "refresh_after_nanos",
                &self.refresh_after_nanos.load(Ordering::Relaxed),
            )
            .field(
                "cold_failure_generation",
                &self.cold_failure_generation.load(Ordering::Relaxed),
            )
            .finish_non_exhaustive()
    }
}

impl SystemProxyConfigCache {
    fn new(
        current: Option<SystemProxyConfig>,
        ttl: Duration,
        reader: SystemProxyConfigReader,
    ) -> Self {
        let epoch = Instant::now();
        let refresh_after_nanos = if current.is_some() {
            duration_nanos(ttl)
        } else {
            0
        };
        Self {
            current: ArcSwapOption::from(current.map(Arc::new)),
            ttl,
            epoch,
            refresh_after_nanos: AtomicU64::new(refresh_after_nanos),
            cold_failure_generation: AtomicU64::new(0),
            refresh_lock: tokio::sync::Mutex::new(()),
            reader,
        }
    }

    fn cached(&self) -> Option<Arc<SystemProxyConfig>> {
        self.current.load_full()
    }

    fn is_fresh(&self, now: u64) -> bool {
        now < self.refresh_after_nanos.load(Ordering::Acquire)
    }

    fn schedule_next_refresh(&self) {
        let now = duration_nanos(self.epoch.elapsed());
        self.refresh_after_nanos.store(
            now.saturating_add(duration_nanos(self.ttl)),
            Ordering::Release,
        );
    }

    async fn refresh(
        &self,
        stale: Option<Arc<SystemProxyConfig>>,
        load_error_policy: &LoadErrorPolicy,
    ) -> Result<Arc<SystemProxyConfig>, BoxError> {
        match self.reader.serve(()).await {
            Ok(config) => {
                let config = Arc::new(config);
                self.current.store(Some(config.clone()));
                self.schedule_next_refresh();
                Ok(config)
            }
            Err(error) => {
                let Some(stale) = stale else {
                    load_error_policy.handle(error)?;
                    let config = Arc::new(SystemProxyConfig::default());
                    self.current.store(Some(config.clone()));
                    self.schedule_next_refresh();
                    return Ok(config);
                };
                self.schedule_next_refresh();
                if let Err(error) = load_error_policy.handle(error) {
                    rama_core::telemetry::tracing::warn!(
                        error = %error,
                        "failed to refresh system proxy configuration; retaining prior snapshot"
                    );
                }
                Ok(stale)
            }
        }
    }

    async fn snapshot(
        &self,
        refresh_enabled: bool,
        refresh_request: &RefreshRequest,
        load_error_policy: &LoadErrorPolicy,
    ) -> Result<Arc<SystemProxyConfig>, BoxError> {
        let refresh_requested = refresh_request.is_pending();
        let current = self.current.load_full();
        let now = duration_nanos(self.epoch.elapsed());
        if let Some(current) = current
            .as_ref()
            .filter(|_| !refresh_enabled || (!refresh_requested && self.is_fresh(now)))
        {
            return Ok(current.clone());
        }

        if let Some(stale) = current {
            let Ok(_guard) = self.refresh_lock.try_lock() else {
                return Ok(stale);
            };
            let latest = self.current.load_full().unwrap_or(stale);
            let now = duration_nanos(self.epoch.elapsed());
            if !refresh_request.is_pending() && self.is_fresh(now) {
                return Ok(latest);
            }
            let result = self.refresh(Some(latest), load_error_policy).await;
            refresh_request.complete();
            return result;
        }

        // Waiters that observed the same cold state share one failed attempt.
        // A later independent call still retries immediately, but a queue of
        // requests cannot serially repeat one slow platform failure.
        let observed_failure = self.cold_failure_generation.load(Ordering::Acquire);
        let _guard = self.refresh_lock.lock().await;
        if let Some(current) = self.current.load_full() {
            if !refresh_request.is_pending() {
                return Ok(current);
            }
            let result = self.refresh(Some(current), load_error_policy).await;
            refresh_request.complete();
            return result;
        }
        if observed_failure != self.cold_failure_generation.load(Ordering::Acquire) {
            return Err(BoxError::from_static_str(
                "system proxy configuration load failed while this request was waiting",
            ));
        }
        let result = self.refresh(None, load_error_policy).await;
        refresh_request.complete();
        if result.is_err() {
            self.cold_failure_generation.fetch_add(1, Ordering::Release);
        }
        result
    }
}

fn duration_nanos(duration: Duration) -> u64 {
    duration.as_nanos().try_into().unwrap_or(u64::MAX)
}

#[doc(hidden)]
#[derive(Debug, Clone, Copy, Default)]
pub struct SystemProxyPacDisabled;

#[doc(hidden)]
#[derive(Debug, Clone, Copy)]
pub struct SystemProxyPacDisabledResolver;

impl Service<Uri> for SystemProxyPacDisabled {
    type Output = SystemProxyPacDisabledResolver;
    type Error = std::convert::Infallible;

    async fn serve(&self, _uri: Uri) -> Result<Self::Output, Self::Error> {
        Ok(SystemProxyPacDisabledResolver)
    }
}

impl Service<SystemProxyPacRequest> for SystemProxyPacDisabledResolver {
    type Output = Option<ProxyRoutes>;
    type Error = std::convert::Infallible;

    async fn serve(&self, _request: SystemProxyPacRequest) -> Result<Self::Output, Self::Error> {
        Ok(None)
    }
}

/// Apply the operating system's proxy settings to client service inputs.
///
/// Existing [`ProxyRoute`] or [`ProxyRoutes`] extensions win by default. This
/// makes the layer safe to place below explicit CLI/application proxy layers:
/// a common priority chain is [`NoProxyEnvLayer`][crate::client::NoProxyEnvLayer],
/// an explicit option, [`ProxyEnvLayer`][crate::client::ProxyEnvLayer], then
/// this system layer. Use
/// [`with_overwrite`][Self::with_overwrite] only when the system policy must
/// replace a route already chosen by the caller.
///
/// Environment proxy variables are intentionally out of scope. Use
/// [`ProxyEnvLayer`][crate::client::ProxyEnvLayer] for proxy variables and
/// [`NoProxyEnvLayer`][crate::client::NoProxyEnvLayer] for bypass variables.
///
/// Configuration discovery is lazy. macOS, Windows, and Linux install a
/// private native change monitor with the first load; the cache TTL remains a
/// fallback on every platform. Applications can replace that monitor through
/// [`with_config_change_trigger`][Self::with_config_change_trigger], retain
/// TTL-only refreshes through
/// [`without_config_change_trigger`][Self::without_config_change_trigger], or
/// make the first snapshot immutable through
/// [`with_config_refresh(false)`][Self::with_config_refresh].
///
/// Fixed proxies and bypass decisions publish one [`ProxyRoute`]. PAC verdicts
/// retain their ordered [`ProxyRoutes`] plan even when it contains one entry.
/// Compose [`ProxyRoutesLayer`][crate::client::ProxyRoutesLayer] after all route
/// selectors and before route-aware middleware.
///
/// A configured PAC URI is used only after a service is supplied through
/// [`with_pac_service`][Self::with_pac_service]. Without one the layer uses a
/// fixed proxy from the same system snapshot when available, or leaves the
/// request unchanged. Factory, fetch, or evaluation errors fail the request
/// instead of silently bypassing the system proxy.
#[derive(Clone)]
pub struct SystemProxyLayer<P = SystemProxyPacDisabled> {
    config: Arc<SystemProxyConfigCache>,
    refresh: SystemProxyConfigRefresh,
    load_error_policy: LoadErrorPolicy,
    pac: P,
    pac_enabled: bool,
    overwrite: bool,
}

impl<P: fmt::Debug> fmt::Debug for SystemProxyLayer<P> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SystemProxyLayer")
            .field("config", &self.config)
            .field("refresh", &self.refresh)
            .field("load_error_policy", &self.load_error_policy)
            .field("pac", &self.pac)
            .field("pac_enabled", &self.pac_enabled)
            .field("overwrite", &self.overwrite)
            .finish()
    }
}

impl SystemProxyLayer {
    /// Create a lazy system-proxy layer without reading platform settings.
    ///
    /// The first unrouted request loads the settings. Call
    /// [`warm_up`][Self::warm_up] to perform that asynchronous load eagerly.
    #[must_use]
    pub fn new() -> Self {
        Self::new_with_ttl_and_invalid_bypass_rule_policy(
            DEFAULT_SYSTEM_PROXY_CONFIG_TTL,
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
    }

    /// Create a lazy system-proxy layer with a custom cache TTL.
    #[must_use]
    pub fn new_with_ttl(ttl: Duration) -> Self {
        Self::new_with_ttl_and_invalid_bypass_rule_policy(
            ttl,
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
    }

    /// Create a lazy layer with explicit refresh and bypass-rule policies.
    #[must_use]
    pub fn new_with_ttl_and_invalid_bypass_rule_policy(
        ttl: Duration,
        policy: SystemProxyInvalidBypassRulePolicy,
    ) -> Self {
        Self::new_with_reader(ttl, system_proxy_config_reader(policy))
    }

    fn new_with_reader(ttl: Duration, reader: SystemProxyConfigReader) -> Self {
        Self {
            config: Arc::new(SystemProxyConfigCache::new(None, ttl, reader)),
            refresh: SystemProxyConfigRefresh::default(),
            load_error_policy: LoadErrorPolicy::Reject,
            pac: SystemProxyPacDisabled,
            pac_enabled: false,
            overwrite: false,
        }
    }

    /// Create a layer from a cached operating-system proxy snapshot.
    ///
    /// The supplied snapshot is used immediately and refreshed from the
    /// operating system after [`DEFAULT_SYSTEM_PROXY_CONFIG_TTL`]. Use
    /// [`try_from_system`][Self::try_from_system] to start with a fresh read.
    #[must_use]
    pub fn from_cached(config: SystemProxyConfig) -> Self {
        Self::from_cached_with_invalid_bypass_rule_policy(
            config,
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
    }

    /// Create a layer from a cached operating-system proxy snapshot with an
    /// explicit invalid bypass-rule policy for subsequent refreshes.
    #[must_use]
    pub fn from_cached_with_invalid_bypass_rule_policy(
        config: SystemProxyConfig,
        policy: SystemProxyInvalidBypassRulePolicy,
    ) -> Self {
        Self::from_cached_with_reader(
            config,
            DEFAULT_SYSTEM_PROXY_CONFIG_TTL,
            system_proxy_config_reader(policy),
        )
    }

    fn from_cached_with_reader(
        config: SystemProxyConfig,
        ttl: Duration,
        reader: SystemProxyConfigReader,
    ) -> Self {
        Self {
            config: Arc::new(SystemProxyConfigCache::new(Some(config), ttl, reader)),
            refresh: SystemProxyConfigRefresh::default(),
            load_error_policy: LoadErrorPolicy::Reject,
            pac: SystemProxyPacDisabled,
            pac_enabled: false,
            overwrite: false,
        }
    }

    /// Create a layer and asynchronously warm its system proxy snapshot.
    pub async fn try_from_system() -> Result<Self, BoxError> {
        Self::try_from_system_with_invalid_bypass_rule_policy(
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
        .await
    }

    /// Create a layer from the current operating system proxy settings using
    /// an explicit invalid bypass-rule policy.
    pub async fn try_from_system_with_invalid_bypass_rule_policy(
        policy: SystemProxyInvalidBypassRulePolicy,
    ) -> Result<Self, BoxError> {
        Self::try_from_system_with_ttl_and_invalid_bypass_rule_policy(
            DEFAULT_SYSTEM_PROXY_CONFIG_TTL,
            policy,
        )
        .await
    }

    /// Create a refreshing layer from the current operating system settings.
    ///
    /// The initial read is awaited. Once `ttl` has elapsed, one request awaits
    /// the refresh while concurrent requests continue using the prior
    /// snapshot. A failed refresh retains that snapshot and is retried after
    /// another `ttl` interval.
    pub async fn try_from_system_with_ttl(ttl: Duration) -> Result<Self, BoxError> {
        Self::try_from_system_with_ttl_and_invalid_bypass_rule_policy(
            ttl,
            SystemProxyInvalidBypassRulePolicy::Ignore,
        )
        .await
    }

    /// Create a refreshing layer with explicit refresh and invalid bypass-rule
    /// policies.
    pub async fn try_from_system_with_ttl_and_invalid_bypass_rule_policy(
        ttl: Duration,
        policy: SystemProxyInvalidBypassRulePolicy,
    ) -> Result<Self, BoxError> {
        Self::try_from_system_with_reader(ttl, system_proxy_config_reader(policy)).await
    }

    async fn try_from_system_with_reader(
        ttl: Duration,
        reader: SystemProxyConfigReader,
    ) -> Result<Self, BoxError> {
        let layer = Self::new_with_reader(ttl, reader);
        layer.warm_up().await?;
        Ok(layer)
    }
}

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

impl<P> SystemProxyLayer<P> {
    /// Load and return the current operating system proxy configuration.
    ///
    /// The first call performs lazy discovery. A native or custom change
    /// trigger can request an early refresh. Once the cache TTL expires, one
    /// caller awaits a refresh while concurrent callers use the stale
    /// snapshot.
    pub async fn config(&self) -> Result<Arc<SystemProxyConfig>, BoxError> {
        let refresh_request = self.refresh.requested().await;
        self.config
            .snapshot(
                self.refresh.enabled,
                &refresh_request,
                &self.load_error_policy,
            )
            .await
    }

    /// Return the cached snapshot without loading or refreshing it.
    #[must_use]
    pub fn cached_config(&self) -> Option<Arc<SystemProxyConfig>> {
        self.config.cached()
    }

    /// Asynchronously populate the cache before serving requests.
    pub async fn warm_up(&self) -> Result<(), BoxError> {
        self.config().await.map(drop)
    }

    /// Supply a PAC resolver factory.
    ///
    /// The factory can be consulted for every request selected for PAC
    /// evaluation. Implementations should therefore reuse resolver state for
    /// the same script URI. Factory and resolver errors fail the request.
    #[must_use]
    pub fn with_pac_service<Q>(self, pac: Q) -> SystemProxyLayer<Q> {
        SystemProxyLayer {
            config: self.config,
            refresh: self.refresh,
            load_error_policy: self.load_error_policy,
            pac,
            pac_enabled: true,
            overwrite: self.overwrite,
        }
    }

    generate_set_and_with! {
        /// Handle system configuration load errors with a sink.
        ///
        /// By default, an initial discovery failure rejects the request. With
        /// this opt-in policy, the error is sent to `error_sink` and an empty
        /// snapshot is cached for the configured TTL. Refresh failures retain
        /// the previous snapshot and are sent to the same sink.
        pub fn load_error_sink(mut self, error_sink: impl ErrorSink) -> Self {
            self.load_error_policy = LoadErrorPolicy::Handle(Arc::new(error_sink));
            self
        }
    }

    generate_set_and_with! {
        /// Enable periodic and change-triggered configuration refreshes.
        ///
        /// Disabling refresh keeps the first loaded snapshot immutable. A
        /// layer created with [`from_cached`][Self::from_cached] therefore
        /// performs no operating-system reads when refresh is disabled.
        pub fn config_refresh(mut self, config_refresh: bool) -> Self {
            self.refresh.enabled = config_refresh;
            self
        }
    }

    generate_set_and_with! {
        /// Replace the default platform configuration-change trigger.
        ///
        /// The service is polled before configuration access. Returning `true`
        /// requests an immediate refresh of an existing snapshot; returning
        /// `false` leaves the TTL as the fallback. Trigger errors are sent to the
        /// configured error sink and also fall back to the TTL.
        pub fn config_change_trigger(
            mut self,
            trigger: impl Service<(), Output = bool, Error: Into<BoxError>>,
        ) -> Self {
            self.refresh.trigger = Some(SystemProxyConfigChangeTrigger::Custom(BoxService::new(
                IntoBoxErrorService(trigger),
            )));
            self
        }
    }

    /// Disable early change notifications while retaining TTL refreshes.
    #[must_use]
    pub fn without_config_change_trigger(mut self) -> Self {
        self.refresh.trigger = None;
        self
    }

    /// Disable early change notifications while retaining TTL refreshes.
    pub fn unset_config_change_trigger(&mut self) -> &mut Self {
        self.refresh.trigger = None;
        self
    }

    generate_set_and_with! {
        /// Replace the sink for configuration-change trigger errors.
        pub fn config_change_trigger_error_sink(
            mut self,
            error_sink: impl ErrorSink,
        ) -> Self {
            self.refresh.trigger_error_sink = Arc::new(error_sink);
            self
        }
    }

    generate_set_and_with! {
        /// Replace an existing route decision (defaults to `false`).
        pub fn overwrite(mut self, overwrite: bool) -> Self {
            self.overwrite = overwrite;
            self
        }
    }
}

impl<S, P> Layer<S> for SystemProxyLayer<P>
where
    P: Clone,
{
    type Service = SystemProxyService<S, P>;

    fn layer(&self, inner: S) -> Self::Service {
        SystemProxyService {
            inner,
            layer: self.clone(),
        }
    }

    fn into_layer(self, inner: S) -> Self::Service {
        SystemProxyService { inner, layer: self }
    }
}

/// See [`SystemProxyLayer`].
#[derive(Debug, Clone)]
pub struct SystemProxyService<S, P = SystemProxyPacDisabled> {
    inner: S,
    layer: SystemProxyLayer<P>,
}

impl<S, P> SystemProxyService<S, P> {
    /// Borrow the wrapped service.
    #[must_use]
    pub const fn inner(&self) -> &S {
        &self.inner
    }

    /// Mutably borrow the wrapped service.
    #[must_use]
    pub fn inner_mut(&mut self) -> &mut S {
        &mut self.inner
    }

    /// Consume this service and return the wrapped service.
    #[must_use]
    pub fn into_inner(self) -> S {
        self.inner
    }
}

impl<S, P, Input> Service<Input> for SystemProxyService<S, P>
where
    S: Service<Input, Error: Into<BoxError>>,
    P: SystemProxyPacService,
    Input: UriInputExt + AuthorityInputExt + ProtocolInputExt + ExtensionsRef + Send + 'static,
{
    type Output = S::Output;
    type Error = BoxError;

    async fn serve(&self, input: Input) -> Result<Self::Output, Self::Error> {
        if !self.layer.overwrite && is_already_routed(&input) {
            return self.inner.serve(input).await.map_err(Into::into);
        }
        let config = self.layer.config().await?;
        if config.is_empty() {
            return self.inner.serve(input).await.map_err(Into::into);
        }

        let mut normalized_uri = None;
        let decision = if self.layer.pac_enabled && config.pac_uri().is_some() {
            let uri = absolute_uri(&input)?;
            let decision = config.decision(&uri);
            normalized_uri = Some(uri);
            decision
        } else {
            let protocol = request_protocol(&input);
            let authority = input
                .uri()
                .authority()
                .map(|authority| authority.into_owned().address)
                .or_else(|| input.authority());
            if let Some(authority) = authority {
                config
                    .fixed_route_for(
                        Some(&protocol),
                        authority.host.view(),
                        authority.port_u16().or_else(|| protocol.default_port()),
                    )
                    .map(SystemProxyDecision::Route)
                    .unwrap_or(SystemProxyDecision::None)
            } else {
                rama_core::telemetry::tracing::debug!(
                    "fixed system proxy cannot route an input without an authority"
                );
                SystemProxyDecision::None
            }
        };
        match decision {
            SystemProxyDecision::Pac(pac_uri) => {
                let Some(uri) = normalized_uri else {
                    return Err(BoxError::from_static_str(
                        "system PAC decision is missing its normalized request URI",
                    ));
                };
                let resolver = self
                    .layer
                    .pac
                    .serve(pac_uri)
                    .await
                    .context("create system PAC resolver")?;
                match resolver
                    .serve(SystemProxyPacRequest::new(
                        input.extensions().clone(),
                        uri.clone(),
                    )?)
                    .await
                    .context("resolve system PAC routes")?
                {
                    Some(routes) => {
                        input.extensions().insert(routes);
                    }
                    None => {
                        if let Some(route) = config.fixed_route(&uri) {
                            input.extensions().insert(route);
                        }
                    }
                }
            }
            SystemProxyDecision::Route(route) => {
                input.extensions().insert(route);
            }
            SystemProxyDecision::None => {}
        }
        self.inner.serve(input).await.map_err(Into::into)
    }
}

pub(super) fn absolute_uri<I>(input: &I) -> Result<Uri, BoxError>
where
    I: UriInputExt + AuthorityInputExt + ProtocolInputExt,
{
    let uri = input.uri();
    let protocol = request_protocol(input);
    proxy_request_uri(uri, input.authority(), protocol)
}

pub(super) fn request_protocol<I>(input: &I) -> Protocol
where
    I: UriInputExt + ProtocolInputExt,
{
    input
        .uri()
        .scheme()
        .cloned()
        // Authority-form is the request-target form of CONNECT. The tunnel is
        // opaque and overwhelmingly TLS, so match the HTTP PAC layer and show
        // it as HTTPS regardless of the named port.
        .or_else(|| input.uri().authority().map(|_| Protocol::HTTPS))
        .or_else(|| input.protocol().cloned())
        .unwrap_or(Protocol::HTTP)
}

/// Normalize a request target for fixed-proxy selection and PAC evaluation.
///
/// The result is absolute, has a root path when no path was supplied, and
/// omits the protocol's default port. The URI's own authority wins over the
/// fallback authority supplied by request metadata.
pub fn proxy_request_uri(
    uri: &Uri,
    fallback_authority: Option<HostWithOptPort>,
    protocol: Protocol,
) -> Result<Uri, BoxError> {
    let authority = uri
        .authority()
        .map(|authority| authority.into_owned().address)
        .or(fallback_authority)
        .ok_or_else(|| BoxError::from_static_str("request has no resolvable authority"))?
        .without_default_port_for(Some(&protocol));

    let mut uri = if uri.is_asterisk() {
        Uri::from_authority(protocol, authority)
    } else {
        uri.clone()
            .with_authority(Authority::from(authority))
            .with_scheme(protocol)
    };
    uri.ensure_path_or_root();
    Ok(uri)
}

pub(super) fn is_already_routed(input: &impl ExtensionsRef) -> bool {
    input.extensions().contains::<ProxyRoute>() || input.extensions().contains::<ProxyRoutes>()
}

#[cfg(any(
    test,
    target_vendor = "apple",
    target_os = "android",
    target_os = "linux",
    target_os = "freebsd",
    target_os = "netbsd",
    target_os = "openbsd",
    target_os = "dragonfly"
))]
pub(super) fn proxy_address(
    protocol: Protocol,
    host: impl AsRef<str>,
    port: u16,
) -> Result<ProxyAddress, BoxError> {
    let value = host.as_ref().trim();
    let host = match Host::try_from(value) {
        Ok(host) => host,
        Err(error) if value.contains("://") => value
            .parse::<Uri>()
            .context("parse system proxy host URI")?
            .host()
            .map(|host| host.into_owned())
            .ok_or(error)
            .context("parse system proxy host")?,
        Err(error) => return Err(error).context("parse system proxy host"),
    };
    Ok(ProxyAddress {
        protocol: Some(protocol),
        address: HostWithPort::new(host, port),
        credential: None,
    })
}

#[cfg(test)]
mod tests {
    use std::convert::Infallible;

    use parking_lot::Mutex;
    use rama_core::{
        extensions::{Extension, FromExtensions},
        service::service_fn,
    };

    use super::*;

    #[derive(Debug, Clone, Extension)]
    struct Marker(&'static str);

    #[derive(FromExtensions)]
    enum RecordedProxyDecision {
        Route(Arc<ProxyRoute>),
        Routes(Arc<ProxyRoutes>),
    }

    #[derive(Debug, Clone)]
    struct TestInput {
        uri: Uri,
        protocol: Option<Protocol>,
        authority: Option<crate::address::HostWithOptPort>,
        extensions: Extensions,
    }

    impl TestInput {
        fn new(uri: &str) -> Self {
            Self {
                uri: uri.parse().unwrap(),
                protocol: None,
                authority: None,
                extensions: Extensions::new(),
            }
        }

        fn origin_form(uri: &str, protocol: Protocol, authority: &str) -> Self {
            Self {
                uri: uri.parse().unwrap(),
                protocol: Some(protocol),
                authority: Some(authority.parse().unwrap()),
                extensions: Extensions::new(),
            }
        }

        fn authority_form(authority: &str) -> Self {
            Self {
                uri: Uri::parse_authority_form(authority).unwrap(),
                protocol: None,
                authority: None,
                extensions: Extensions::new(),
            }
        }
    }

    impl UriInputExt for TestInput {
        fn uri(&self) -> &Uri {
            &self.uri
        }
    }

    impl AuthorityInputExt for TestInput {
        fn authority(&self) -> Option<crate::address::HostWithOptPort> {
            self.authority.clone().or_else(|| {
                self.uri
                    .authority()
                    .map(|authority| authority.into_owned().address)
            })
        }
    }

    impl ProtocolInputExt for TestInput {
        fn protocol(&self) -> Option<&Protocol> {
            self.protocol.as_ref().or_else(|| self.uri.scheme())
        }
    }

    impl ExtensionsRef for TestInput {
        fn extensions(&self) -> &Extensions {
            &self.extensions
        }
    }

    fn proxy(protocol: Protocol, host: &'static str, port: u16) -> ProxyAddress {
        proxy_address(protocol, host, port).unwrap()
    }

    fn recorder() -> (
        impl Service<TestInput, Output = (), Error = Infallible> + Clone,
        Arc<Mutex<Vec<Option<ProxyRoutes>>>>,
    ) {
        let seen = Arc::new(Mutex::new(Vec::new()));
        let service = service_fn({
            let seen = seen.clone();
            move |input: TestInput| {
                let routes = match RecordedProxyDecision::from_extensions(&input.extensions) {
                    Some(RecordedProxyDecision::Route(route)) => {
                        Some(ProxyRoutes::from(route.as_ref().clone()))
                    }
                    Some(RecordedProxyDecision::Routes(routes)) => Some(routes.as_ref().clone()),
                    None => None,
                };
                seen.lock().push(routes);
                async { Ok::<_, Infallible>(()) }
            }
        });
        (service, seen)
    }

    #[tokio::test]
    async fn fixed_proxies_are_selected_by_destination_scheme() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "http.proxy", 8080))
            .with_https_proxy(proxy(Protocol::HTTP, "https.proxy", 8443));
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config).into_layer(inner);

        service
            .serve(TestInput::new("http://example.com/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("https://example.com/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("ws://example.com/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("wss://example.com/"))
            .await
            .unwrap();

        let seen = seen.lock();
        assert_eq!(
            seen[0].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "http.proxy"
        );
        assert_eq!(
            seen[1].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "https.proxy"
        );
        assert_eq!(
            seen[2].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "http.proxy"
        );
        assert_eq!(
            seen[3].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "https.proxy"
        );
    }

    #[tokio::test]
    async fn fixed_and_bypass_decisions_publish_singular_routes() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let service = SystemProxyLayer::from_cached(config).into_layer(service_fn(
            async |input: TestInput| Ok::<_, Infallible>(input),
        ));

        let proxied = service
            .serve(TestInput::new("http://example.com/"))
            .await
            .unwrap();
        assert_eq!(
            proxied
                .extensions
                .get_ref::<ProxyRoute>()
                .and_then(ProxyRoute::proxy_address)
                .map(|address| address.address.host.to_string()),
            Some("system.proxy".to_owned())
        );
        assert!(!proxied.extensions.contains::<ProxyRoutes>());

        let bypassed = service
            .serve(TestInput::new("http://localhost/"))
            .await
            .unwrap();
        assert_eq!(
            bypassed.extensions.get_ref::<ProxyRoute>(),
            Some(&ProxyRoute::Direct)
        );
        assert!(!bypassed.extensions.contains::<ProxyRoutes>());
    }

    #[tokio::test]
    async fn system_decisions_override_configured_route_defaults() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let service = SystemProxyLayer::from_cached(config).into_layer(
            crate::client::ProxyRoutesLayer::with_routes(ProxyRoute::Proxy(proxy(
                Protocol::HTTP,
                "default.proxy",
                8080,
            )))
            .into_layer(service_fn(async |input: TestInput| {
                Ok::<_, Infallible>(input)
            })),
        );

        let proxied = service
            .serve(TestInput::new("http://example.com/"))
            .await
            .unwrap();
        assert_eq!(
            proxied
                .extensions
                .get_ref::<ProxyRoute>()
                .and_then(ProxyRoute::proxy_address)
                .map(|address| address.address.host.to_string()),
            Some("system.proxy".to_owned())
        );

        let bypassed = service
            .serve(TestInput::new("http://localhost/"))
            .await
            .unwrap();
        assert_eq!(
            bypassed.extensions.get_ref::<ProxyRoute>(),
            Some(&ProxyRoute::Direct)
        );
    }

    #[tokio::test]
    async fn socks_is_the_scheme_independent_fallback() {
        let config = SystemProxyConfig::default().with_socks5_proxy(proxy(
            Protocol::SOCKS5,
            "socks.proxy",
            1080,
        ));
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config).into_layer(inner);

        service
            .serve(TestInput::new("https://example.com/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("ftp://example.com/file"))
            .await
            .unwrap();

        let seen = seen.lock();
        for routes in seen.iter() {
            let address = routes.as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap();
            assert_eq!(address.protocol, Some(Protocol::SOCKS5));
        }
    }

    #[tokio::test]
    async fn scheme_specific_proxy_does_not_capture_other_protocols() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "http.proxy", 8080))
            .with_bypass(["example.com"]);
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(TestInput::new("ftp://example.com/file"))
            .await
            .unwrap();

        assert!(seen.lock()[0].is_none());
    }

    #[tokio::test]
    async fn empty_config_does_not_require_routing_metadata() {
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(SystemProxyConfig::default())
            .into_layer(inner)
            .serve(TestInput::new("/relative"))
            .await
            .unwrap();

        assert!(seen.lock()[0].is_none());
    }

    #[tokio::test]
    async fn active_fixed_config_passes_input_without_an_authority() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(TestInput::new("/relative"))
            .await
            .unwrap();

        assert!(seen.lock()[0].is_none());
    }

    #[tokio::test]
    async fn input_without_a_protocol_defaults_to_http() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let (inner, seen) = recorder();
        let mut input = TestInput::new("/relative");
        input.authority = Some("example.com".parse().unwrap());

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(input)
            .await
            .unwrap();

        assert!(matches!(
            seen.lock()[0].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
    }

    #[tokio::test]
    async fn existing_route_wins_unless_overwrite_is_enabled() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let (inner, seen) = recorder();
        let request = TestInput::new("http://example.com/");
        request.extensions.insert(ProxyRoutes::from(proxy(
            Protocol::HTTP,
            "explicit.proxy",
            9000,
        )));

        SystemProxyLayer::from_cached(config.clone())
            .into_layer(inner.clone())
            .serve(request.clone())
            .await
            .unwrap();
        SystemProxyLayer::from_cached(config)
            .with_overwrite(true)
            .into_layer(inner)
            .serve(request)
            .await
            .unwrap();

        let seen = seen.lock();
        let hosts: Vec<_> = seen
            .iter()
            .map(|routes| {
                routes.as_ref().unwrap().as_slice()[0]
                    .proxy_address()
                    .unwrap()
                    .address
                    .host
                    .to_str()
                    .into_owned()
            })
            .collect();
        assert_eq!(hosts, ["explicit.proxy", "system.proxy"]);
    }

    #[tokio::test]
    async fn overwrite_route_takes_priority_over_an_existing_route() {
        let config = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "system.proxy",
            8080,
        ));
        let request = TestInput::new("http://example.com/");
        request.extensions.insert(ProxyRoute::Direct);
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .with_overwrite(true)
            .into_layer(inner)
            .serve(request)
            .await
            .unwrap();

        let seen = seen.lock();
        let routes = seen[0].as_ref().unwrap();
        assert_eq!(
            routes.as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "system.proxy"
        );
    }

    #[tokio::test]
    async fn pac_receives_full_uri_and_cloned_extensions() {
        let pac_uri: Uri = "http://config.example/proxy.pac".parse().unwrap();
        let factory_seen = Arc::new(Mutex::new(Vec::new()));
        let resolver_seen = Arc::new(Mutex::new(Vec::new()));
        let factory = service_fn({
            let factory_seen = factory_seen.clone();
            let resolver_seen = resolver_seen.clone();
            move |uri: Uri| {
                factory_seen.lock().push(uri);
                let resolver_seen = resolver_seen.clone();
                async move {
                    Ok::<_, Infallible>(service_fn(move |request: SystemProxyPacRequest| {
                        resolver_seen.lock().push((
                            request.uri.clone(),
                            request.extensions().get_ref::<Marker>().cloned(),
                        ));
                        async move {
                            Ok::<_, Infallible>(Some(ProxyRoutes::from(proxy(
                                Protocol::HTTP,
                                "pac.proxy",
                                8080,
                            ))))
                        }
                    }))
                }
            }
        });
        let config = SystemProxyConfig::default().with_pac_uri(pac_uri.clone());
        let request = TestInput::new("https://example.com/private?q=1");
        request.extensions.insert(Marker("kept"));
        let inner = service_fn(async |input: TestInput| Ok::<_, Infallible>(input));

        let output = SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(request)
            .await
            .unwrap();

        assert_eq!(factory_seen.lock().as_slice(), [pac_uri]);
        let resolved = resolver_seen.lock();
        assert_eq!(resolved[0].0.to_string(), "https://example.com/private?q=1");
        assert_eq!(resolved[0].1.as_ref().unwrap().0, "kept");
        assert_eq!(
            output
                .extensions
                .get_ref::<ProxyRoutes>()
                .unwrap()
                .as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "pac.proxy"
        );
        assert!(output.extensions.get_ref::<ProxyRoute>().is_none());
    }

    #[tokio::test]
    async fn pac_factory_errors_fail_the_request_with_context() {
        let factory = service_fn(|_uri: Uri| async {
            Err::<SystemProxyPacDisabledResolver, _>(std::io::Error::other("PAC fetch failed"))
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let (inner, seen) = recorder();

        let error = SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(TestInput::new("https://example.com/"))
            .await
            .unwrap_err();

        assert!(error.to_string().contains("create system PAC resolver"));
        assert!(seen.lock().is_empty());
    }

    #[tokio::test]
    async fn pac_resolver_errors_fail_the_request_with_context() {
        let factory = service_fn(|_uri: Uri| async {
            Ok::<_, Infallible>(service_fn(|_request: SystemProxyPacRequest| async {
                Err::<Option<ProxyRoutes>, _>(std::io::Error::other("PAC evaluation failed"))
            }))
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let (inner, seen) = recorder();

        let error = SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(TestInput::new("https://example.com/"))
            .await
            .unwrap_err();

        assert!(error.to_string().contains("resolve system PAC routes"));
        assert!(seen.lock().is_empty());
    }

    #[tokio::test]
    async fn pac_receives_an_absolute_uri_for_origin_form_input() {
        let received = Arc::new(Mutex::new(None));
        let factory = service_fn({
            let received = received.clone();
            move |_uri: Uri| {
                let received = received.clone();
                async move {
                    Ok::<_, Infallible>(service_fn(move |request: SystemProxyPacRequest| {
                        *received.lock() = Some(request.uri);
                        async { Ok::<_, Infallible>(None) }
                    }))
                }
            }
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let (inner, _) = recorder();

        SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(TestInput::origin_form(
                "/private?q=1",
                Protocol::HTTPS,
                "example.com:8443",
            ))
            .await
            .unwrap();

        assert_eq!(
            received.lock().as_ref().unwrap().to_string(),
            "https://example.com:8443/private?q=1"
        );
    }

    #[tokio::test]
    async fn pac_normalizes_default_ports_with_an_unboxed_resolver() {
        let factory_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let received = Arc::new(Mutex::new(Vec::new()));
        let factory = service_fn({
            let factory_calls = factory_calls.clone();
            let received = received.clone();
            move |_uri: Uri| {
                factory_calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let received = received.clone();
                async move {
                    Ok::<_, Infallible>(service_fn(move |request: SystemProxyPacRequest| {
                        received.lock().push(request.uri);
                        async { Ok::<_, Infallible>(None) }
                    }))
                }
            }
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let (inner, _) = recorder();
        let service = SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner);

        for input in [
            TestInput::new("http://example.com:80/path"),
            TestInput::new("https://example.com:443/"),
            TestInput::new("http://example.com:8080/"),
            TestInput::authority_form("example.com:443"),
        ] {
            service.serve(input).await.unwrap();
        }

        assert_eq!(factory_calls.load(std::sync::atomic::Ordering::Relaxed), 4);
        assert_eq!(
            received
                .lock()
                .iter()
                .map(ToString::to_string)
                .collect::<Vec<_>>(),
            [
                "http://example.com/path",
                "https://example.com/",
                "http://example.com:8080/",
                "https://example.com/",
            ]
        );
    }

    #[tokio::test]
    async fn authority_form_selects_the_https_proxy() {
        let config = SystemProxyConfig::default().with_https_proxy(proxy(
            Protocol::HTTP,
            "https.proxy",
            8443,
        ));
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(TestInput::authority_form("example.com:443"))
            .await
            .unwrap();

        let routes = seen.lock();
        assert_eq!(
            routes[0].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "https.proxy"
        );
    }

    #[tokio::test]
    async fn pac_without_a_service_leaves_the_request_undecided() {
        let config = SystemProxyConfig::default()
            .with_pac_uri("http://config.example/proxy.pac".parse().unwrap());
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(TestInput::new("/relative"))
            .await
            .unwrap();

        assert!(seen.lock()[0].is_none());
    }

    #[tokio::test]
    async fn pac_without_a_service_uses_a_fixed_proxy_fallback() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "fixed.proxy", 8080))
            .with_pac_uri("http://config.example/proxy.pac".parse().unwrap());
        let (inner, seen) = recorder();

        SystemProxyLayer::from_cached(config)
            .into_layer(inner)
            .serve(TestInput::new("http://example.com/"))
            .await
            .unwrap();

        assert_eq!(
            seen.lock()[0].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "fixed.proxy"
        );
    }

    #[tokio::test]
    async fn active_pac_requires_a_resolvable_authority() {
        let factory = service_fn(|_uri: Uri| async {
            Ok::<_, Infallible>(service_fn(|_request| async { Ok::<_, Infallible>(None) }))
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("http://config.example/proxy.pac".parse().unwrap());
        let (inner, _) = recorder();

        SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(TestInput::new("/relative"))
            .await
            .unwrap_err();
    }

    #[tokio::test]
    async fn singular_route_also_prevents_pac_lookup() {
        let factory_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let factory = service_fn({
            let factory_calls = factory_calls.clone();
            move |_uri: Uri| {
                factory_calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    Ok::<_, Infallible>(service_fn(|_request| async { Ok::<_, Infallible>(None) }))
                }
            }
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let request = TestInput::new("https://example.com/");
        request.extensions.insert(ProxyRoute::Direct);
        let (inner, _) = recorder();

        SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner)
            .serve(request)
            .await
            .unwrap();

        assert_eq!(factory_calls.load(std::sync::atomic::Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn an_existing_route_prevents_system_config_refresh() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        "system.proxy",
                        8080,
                    )))
                }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::ZERO, reader)
            .await
            .unwrap();
        let request = TestInput::new("http://example.com/");
        request.extensions.insert(ProxyRoute::Direct);
        let (inner, _) = recorder();

        layer.into_layer(inner).serve(request).await.unwrap();
        tokio::time::sleep(Duration::from_millis(50)).await;

        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn bypass_and_inverted_bypass_select_direct_routes() {
        let base = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "system.proxy", 8080))
            .with_bypass([".example.com", "default-port.test:80"]);
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(base.clone()).into_layer(inner.clone());
        service
            .serve(TestInput::new("http://api.example.com/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("http://elsewhere.test/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("http://default-port.test/"))
            .await
            .unwrap();

        let inverted =
            SystemProxyLayer::from_cached(base.with_reversed_bypass(true)).into_layer(inner);
        inverted
            .serve(TestInput::new("http://api.example.com/"))
            .await
            .unwrap();
        inverted
            .serve(TestInput::new("http://elsewhere.test/"))
            .await
            .unwrap();

        let seen = seen.lock();
        assert!(matches!(
            seen[0].as_ref().unwrap().as_slice(),
            [ProxyRoute::Direct]
        ));
        assert!(matches!(
            seen[1].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
        assert!(matches!(
            seen[2].as_ref().unwrap().as_slice(),
            [ProxyRoute::Direct]
        ));
        assert!(matches!(
            seen[3].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
        assert!(matches!(
            seen[4].as_ref().unwrap().as_slice(),
            [ProxyRoute::Direct]
        ));
    }

    #[tokio::test]
    async fn simple_hostname_bypass_is_opt_in() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "system.proxy", 8080))
            .with_exclude_simple_hostnames(true);
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config).into_layer(inner);

        service
            .serve(TestInput::new("http://printer/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("http://printer.example/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("http://[2001:db8::1]/"))
            .await
            .unwrap();

        let seen = seen.lock();
        assert!(matches!(
            seen[0].as_ref().unwrap().as_slice(),
            [ProxyRoute::Direct]
        ));
        assert!(matches!(
            seen[1].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
        assert!(matches!(
            seen[2].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
    }

    #[tokio::test]
    async fn inverted_simple_hostname_bypass_uses_only_simple_names() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "system.proxy", 8080))
            .with_exclude_simple_hostnames(true)
            .with_reversed_bypass(true);
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config).into_layer(inner);

        service
            .serve(TestInput::new("http://printer/"))
            .await
            .unwrap();
        service
            .serve(TestInput::new("http://printer.example/"))
            .await
            .unwrap();

        let seen = seen.lock();
        assert!(matches!(
            seen[0].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
        assert!(matches!(
            seen[1].as_ref().unwrap().as_slice(),
            [ProxyRoute::Direct]
        ));
    }

    #[tokio::test]
    async fn fixed_system_proxies_implicitly_bypass_loopback() {
        let config = SystemProxyConfig::default()
            .with_http_proxy(proxy(Protocol::HTTP, "system.proxy", 8080))
            .with_bypass(["localhost", "127.0.0.0/8", "::1", "remote.example"])
            .with_reversed_bypass(true);
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config).into_layer(inner);

        for uri in [
            "http://localhost/",
            "http://service.localhost/",
            "http://127.42.0.1/",
            "http://[::1]/",
            "http://remote.example/",
        ] {
            service.serve(TestInput::new(uri)).await.unwrap();
        }

        let seen = seen.lock();
        for routes in &seen[..4] {
            assert!(matches!(
                routes.as_ref().unwrap().as_slice(),
                [ProxyRoute::Direct]
            ));
        }
        assert!(matches!(
            seen[4].as_ref().unwrap().as_slice(),
            [ProxyRoute::Proxy(_)]
        ));
    }

    #[tokio::test]
    async fn pac_is_not_consulted_for_loopback() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let factory = service_fn({
            let calls = calls.clone();
            move |_uri: Uri| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Ok::<_, Infallible>(service_fn(|_request| async {
                        Ok::<_, Infallible>(Some(ProxyRoutes::from(ProxyRoute::Direct)))
                    }))
                }
            }
        });
        let config = SystemProxyConfig::default()
            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap());
        let (inner, seen) = recorder();
        let service = SystemProxyLayer::from_cached(config)
            .with_pac_service(factory)
            .into_layer(inner);

        for uri in [
            "http://localhost/",
            "http://service.localhost/",
            "http://127.42.0.1/",
            "http://[::1]/",
        ] {
            service.serve(TestInput::new(uri)).await.unwrap();
        }

        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0);
        assert!(
            seen.lock()
                .iter()
                .all(|routes| matches!(routes.as_ref().unwrap().as_slice(), [ProxyRoute::Direct]))
        );
    }

    #[test]
    fn platform_bypass_precedence_controls_pac_decision() {
        let pac_uri: Uri = "https://config.example/proxy.pac".parse().unwrap();
        let uri: Uri = "https://bypass.example/".parse().unwrap();
        let mut config = SystemProxyConfig::default()
            .with_pac_uri(pac_uri.clone())
            .with_bypass(["bypass.example"]);

        assert!(matches!(
            config.decision(&uri),
            SystemProxyDecision::Pac(uri) if uri == pac_uri
        ));

        config.bypass_before_pac = true;
        assert!(matches!(
            config.decision(&uri),
            SystemProxyDecision::Route(ProxyRoute::Direct)
        ));
    }

    #[test]
    fn config_accessors_and_public_pac_request_fields_round_trip() {
        let http = proxy(Protocol::HTTP, "http.proxy", 8080);
        let https = proxy(Protocol::HTTP, "https.proxy", 8443);
        let socks = proxy(Protocol::SOCKS5, "socks.proxy", 1080);
        let pac: Uri = "https://config.example/proxy.pac".parse().unwrap();
        let config = SystemProxyConfig::default()
            .with_http_proxy(http.clone())
            .with_https_proxy(https.clone())
            .with_socks5_proxy(socks.clone())
            .with_pac_uri(pac.clone())
            .with_bypass(["localhost"])
            .with_exclude_simple_hostnames(true)
            .with_reversed_bypass(true)
            .with_auto_detect(true);

        assert!(!config.is_empty());
        assert_eq!(config.http_proxy(), Some(&http));
        assert_eq!(config.https_proxy(), Some(&https));
        assert_eq!(config.socks5_proxy(), Some(&socks));
        assert_eq!(config.pac_uri(), Some(&pac));
        assert_eq!(config.bypass().collect::<Vec<_>>(), ["localhost"]);
        assert!(config.exclude_simple_hostnames());
        assert!(config.reversed_bypass());
        assert!(config.auto_detect());

        let extensions = Extensions::new();
        extensions.insert(Marker("parts"));
        let request =
            SystemProxyPacRequest::new(extensions, "http://example.com/path".parse().unwrap())
                .unwrap();
        assert_eq!(
            UriInputExt::uri(&request).to_string(),
            "http://example.com/path"
        );
        assert_eq!(
            ExtensionsRef::extensions(&request)
                .get_ref::<Marker>()
                .unwrap()
                .0,
            "parts"
        );
        assert_eq!(request.extensions.get_ref::<Marker>().unwrap().0, "parts");
        assert_eq!(request.uri.to_string(), "http://example.com/path");
    }

    #[test]
    fn platform_proxy_host_accepts_a_scheme_prefix() {
        let proxy = proxy_address(Protocol::HTTP, "http://proxy.corp", 8080).unwrap();
        assert_eq!(proxy.to_string(), "http://proxy.corp:8080");
    }

    #[cfg(any(target_os = "macos", target_os = "windows", target_os = "linux"))]
    #[tokio::test]
    async fn native_system_proxy_snapshot_can_be_read() {
        SystemProxyConfig::try_from_system().await.unwrap();
    }

    #[test]
    fn every_proxy_source_independently_makes_config_non_empty() {
        assert!(SystemProxyConfig::default().is_empty());
        for config in [
            SystemProxyConfig::default().with_http_proxy(proxy(Protocol::HTTP, "http.proxy", 8080)),
            SystemProxyConfig::default().with_https_proxy(proxy(
                Protocol::HTTP,
                "https.proxy",
                8443,
            )),
            SystemProxyConfig::default().with_socks5_proxy(proxy(
                Protocol::SOCKS5,
                "socks.proxy",
                1080,
            )),
            SystemProxyConfig::default()
                .with_pac_uri("https://config.example/proxy.pac".parse().unwrap()),
            SystemProxyConfig::default().with_auto_detect(true),
        ] {
            assert!(!config.is_empty());
        }
    }

    #[test]
    fn pac_request_rejects_non_absolute_or_hostless_uri() {
        SystemProxyPacRequest::new(Extensions::new(), "/path".parse().unwrap()).unwrap_err();
        SystemProxyPacRequest::new(Extensions::new(), "data:text/plain,x".parse().unwrap())
            .unwrap_err();
    }

    #[test]
    fn bypass_patterns_cover_domains_ports_ip_ranges_and_local_names() {
        for (pattern, scheme, host, port, expected) in [
            ("*", None, "anything.example", None, true),
            ("<local>", None, "printer", None, true),
            ("<local>", None, "printer.example", None, false),
            ("<local>", None, "2001:db8::1", None, false),
            ("192.168.*", None, "192.168.10.20", None, true),
            ("*corp*", None, "api.corp.example", None, true),
            ("*.example.com", None, "api.example.com", None, true),
            ("*.example.com", None, "example.com", None, true),
            (".example.com", None, "api.example.com.", None, true),
            (".example.com", None, "notexample.com", None, false),
            (
                "api.example.com:8443",
                None,
                "api.example.com",
                Some(8443),
                true,
            ),
            (
                "api.example.com:8443",
                None,
                "api.example.com",
                Some(443),
                false,
            ),
            ("10.0.0.0/8", None, "10.2.3.4", None, true),
            ("10.0.0.0/8", None, "11.2.3.4", None, false),
            ("[::1]", None, "::1", None, true),
            ("::1", None, "::1", None, true),
            ("[::1]:8443", None, "::1", Some(8443), true),
            ("[::1]:8443", None, "::1", Some(443), false),
            ("2001:db8::/32", None, "2001:db8::1", None, true),
            (
                "https://secure.example:443",
                Some(Protocol::HTTPS),
                "secure.example",
                Some(443),
                true,
            ),
            (
                "https://secure.example:443",
                Some(Protocol::HTTP),
                "secure.example",
                Some(443),
                false,
            ),
        ] {
            let host = Host::try_from(host).unwrap();
            let host_text = host.to_string();
            assert_eq!(
                BypassRule::compile(pattern).unwrap().matches(
                    scheme.as_ref(),
                    (&host).into(),
                    port,
                ),
                expected,
                "{pattern} {host_text:?} {port:?}"
            );
        }
    }

    #[test]
    fn invalid_bypass_patterns_are_discarded() {
        let config =
            SystemProxyConfig::default().with_bypass(["example.com", ".not a valid domain", ""]);

        assert_eq!(config.bypass().collect::<Vec<_>>(), ["example.com"]);
    }

    #[test]
    fn invalid_bypass_policy_can_reject_an_update_atomically() {
        let mut config = SystemProxyConfig::default().with_bypass(["existing.example"]);

        let error = config
            .try_set_bypass(
                ["replacement.example", ".not a valid domain"],
                SystemProxyInvalidBypassRulePolicy::Reject,
            )
            .unwrap_err();

        assert!(
            error
                .to_string()
                .contains("parse system proxy bypass pattern")
        );
        assert_eq!(config.bypass().collect::<Vec<_>>(), ["existing.example"]);

        config
            .try_set_bypass(
                ["replacement.example", ".not a valid domain"],
                SystemProxyInvalidBypassRulePolicy::Ignore,
            )
            .unwrap();
        assert_eq!(config.bypass().collect::<Vec<_>>(), ["replacement.example"]);
    }

    #[tokio::test]
    async fn lazy_layer_construction_does_not_read_until_warmed() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        "lazy.proxy",
                        8080,
                    )))
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);

        assert!(layer.cached_config().is_none());
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0);

        layer.warm_up().await.unwrap();
        layer.warm_up().await.unwrap();

        let config = layer.cached_config().unwrap();
        assert_eq!(
            config.http_proxy().unwrap().address.host.to_str(),
            "lazy.proxy"
        );
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn change_trigger_refreshes_a_fresh_snapshot() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let changed = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    let host = if call == 0 { "old.proxy" } else { "new.proxy" };
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        host,
                        8080,
                    )))
                }
            }
        }));
        let trigger = service_fn({
            let changed = changed.clone();
            move |()| {
                let changed = changed.swap(false, std::sync::atomic::Ordering::AcqRel);
                async move { Ok::<_, std::convert::Infallible>(changed) }
            }
        });
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader)
            .with_config_change_trigger(trigger);

        let old = layer.config().await.unwrap();
        assert_eq!(old.http_proxy().unwrap().address.host.to_str(), "old.proxy");
        changed.store(true, std::sync::atomic::Ordering::Release);
        let new = layer.config().await.unwrap();
        let still_new = layer.config().await.unwrap();

        assert_eq!(new.http_proxy().unwrap().address.host.to_str(), "new.proxy");
        assert!(Arc::ptr_eq(&new, &still_new));
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn change_during_refresh_remains_pending_for_the_next_request() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let changed = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let refresh_started = Arc::new(tokio::sync::Notify::new());
        let release_refresh = Arc::new(tokio::sync::Notify::new());
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            let refresh_started = refresh_started.clone();
            let release_refresh = release_refresh.clone();
            move |()| {
                let call = reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let refresh_started = refresh_started.clone();
                let release_refresh = release_refresh.clone();
                async move {
                    if call == 0 {
                        refresh_started.notify_one();
                        release_refresh.notified().await;
                    }
                    let host = if call == 0 {
                        "first-refresh.proxy"
                    } else {
                        "second-refresh.proxy"
                    };
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        host,
                        8080,
                    )))
                }
            }
        }));
        let trigger = service_fn({
            let changed = changed.clone();
            move |()| {
                let changed = changed.swap(false, std::sync::atomic::Ordering::AcqRel);
                async move { Ok::<_, Infallible>(changed) }
            }
        });
        let cached = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "cached.proxy",
            8080,
        ));
        let layer =
            SystemProxyLayer::from_cached_with_reader(cached, Duration::from_mins(1), reader)
                .with_config_change_trigger(trigger);

        changed.store(true, std::sync::atomic::Ordering::Release);
        let first_layer = layer.clone();
        let first = tokio::spawn(async move { first_layer.config().await });
        refresh_started.notified().await;

        changed.store(true, std::sync::atomic::Ordering::Release);
        let stale = layer.config().await.unwrap();
        assert_eq!(stale.http_proxy().unwrap().address.host, "cached.proxy");

        release_refresh.notify_one();
        let first = first.await.unwrap().unwrap();
        assert_eq!(
            first.http_proxy().unwrap().address.host,
            "first-refresh.proxy"
        );
        let second = layer.config().await.unwrap();
        assert_eq!(
            second.http_proxy().unwrap().address.host,
            "second-refresh.proxy"
        );
        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn cancelled_triggered_refresh_does_not_consume_the_request() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let changed = Arc::new(std::sync::atomic::AtomicBool::new(true));
        let refresh_started = Arc::new(tokio::sync::Notify::new());
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            let refresh_started = refresh_started.clone();
            move |()| {
                let call = reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let refresh_started = refresh_started.clone();
                async move {
                    if call == 0 {
                        refresh_started.notify_one();
                        std::future::pending::<()>().await;
                    }
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        "refreshed.proxy",
                        8080,
                    )))
                }
            }
        }));
        let trigger = service_fn({
            let changed = changed.clone();
            move |()| {
                let changed = changed.swap(false, std::sync::atomic::Ordering::AcqRel);
                async move { Ok::<_, Infallible>(changed) }
            }
        });
        let cached = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "cached.proxy",
            8080,
        ));
        let layer =
            SystemProxyLayer::from_cached_with_reader(cached, Duration::from_mins(1), reader)
                .with_config_change_trigger(trigger);

        let refresh_layer = layer.clone();
        let refresh = tokio::spawn(async move { refresh_layer.config().await });
        refresh_started.notified().await;
        assert_eq!(
            layer
                .config()
                .await
                .unwrap()
                .http_proxy()
                .unwrap()
                .address
                .host,
            "cached.proxy"
        );
        refresh.abort();
        refresh.await.unwrap_err();

        let refreshed = layer.config().await.unwrap();
        assert_eq!(
            refreshed.http_proxy().unwrap().address.host,
            "refreshed.proxy"
        );
        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn failed_triggered_refresh_is_acknowledged_until_the_ttl() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let changed = Arc::new(std::sync::atomic::AtomicBool::new(true));
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            move |()| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Err::<SystemProxyConfig, BoxError>(std::io::Error::other("offline").into())
                }
            }
        }));
        let trigger = service_fn({
            let changed = changed.clone();
            move |()| {
                let changed = changed.swap(false, std::sync::atomic::Ordering::AcqRel);
                async move { Ok::<_, Infallible>(changed) }
            }
        });
        let layer = SystemProxyLayer::from_cached_with_reader(
            SystemProxyConfig::default(),
            Duration::from_mins(1),
            reader,
        )
        .with_config_change_trigger(trigger);

        layer.config().await.unwrap();
        layer.config().await.unwrap();
        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn disabled_change_trigger_retains_ttl_refresh() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let triggers = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            move |()| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async { Ok::<_, BoxError>(SystemProxyConfig::default()) }
            }
        }));
        let trigger = service_fn({
            let triggers = triggers.clone();
            move |()| {
                triggers.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async { Ok::<_, std::convert::Infallible>(true) }
            }
        });
        let layer = SystemProxyLayer::new_with_reader(Duration::ZERO, reader)
            .with_config_change_trigger(trigger)
            .without_config_change_trigger();

        layer.config().await.unwrap();
        layer.config().await.unwrap();

        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 2);
        assert_eq!(triggers.load(std::sync::atomic::Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn disabled_refresh_makes_a_cached_snapshot_immutable() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            move |()| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async { Ok::<_, BoxError>(SystemProxyConfig::default()) }
            }
        }));
        let cached = SystemProxyConfig::default().with_http_proxy(proxy(
            Protocol::HTTP,
            "cached.proxy",
            8080,
        ));
        let layer = SystemProxyLayer::from_cached_with_reader(cached, Duration::ZERO, reader)
            .with_config_refresh(false);

        for _ in 0..2 {
            assert_eq!(
                layer
                    .config()
                    .await
                    .unwrap()
                    .http_proxy()
                    .unwrap()
                    .address
                    .host
                    .to_str(),
                "cached.proxy"
            );
        }
        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 0);
    }

    #[tokio::test]
    async fn change_trigger_errors_fall_back_to_the_ttl() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let errors = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            move |()| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async { Ok::<_, BoxError>(SystemProxyConfig::default()) }
            }
        }));
        let trigger = service_fn(|()| async {
            Err::<bool, _>(std::io::Error::other("configuration watcher failed"))
        });
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader)
            .with_config_change_trigger(trigger)
            .with_config_change_trigger_error_sink({
                let errors = errors.clone();
                move |_error: BoxError| {
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            });

        layer.config().await.unwrap();
        layer.config().await.unwrap();

        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 1);
        assert_eq!(errors.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn first_request_lazily_loads_and_applies_system_proxy() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        "lazy.proxy",
                        8080,
                    )))
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);
        let (inner, seen) = recorder();
        let service = layer.into_layer(inner);

        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 0);
        service
            .serve(TestInput::new("http://example.com/"))
            .await
            .unwrap();

        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);
        assert_eq!(
            seen.lock()[0].as_ref().unwrap().as_slice()[0]
                .proxy_address()
                .unwrap()
                .address
                .host
                .to_str(),
            "lazy.proxy"
        );
    }

    #[tokio::test]
    async fn concurrent_cold_loads_share_one_async_read() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    tokio::time::sleep(Duration::from_millis(20)).await;
                    Ok::<_, BoxError>(SystemProxyConfig::default())
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);

        let (first, second, third) = tokio::join!(layer.config(), layer.config(), layer.config());

        first.unwrap();
        second.unwrap();
        third.unwrap();
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn failed_cold_load_remains_retryable() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    if call == 0 {
                        Err(std::io::Error::other("temporary platform read failure").into())
                    } else {
                        Ok(SystemProxyConfig::default())
                    }
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);

        layer.config().await.unwrap_err();
        assert!(layer.cached_config().is_none());
        layer.config().await.unwrap();
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn handled_cold_load_error_is_sunk_and_cached_for_the_ttl() {
        let reads = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let errors = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let reads = reads.clone();
            move |()| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async {
                    Err::<SystemProxyConfig, BoxError>(
                        std::io::Error::other("platform read failed").into(),
                    )
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader)
            .with_load_error_sink({
                let errors = errors.clone();
                move |error: BoxError| {
                    assert_eq!(error.to_string(), "platform read failed");
                    errors.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            });

        assert!(layer.config().await.unwrap().is_empty());
        assert!(layer.config().await.unwrap().is_empty());
        assert_eq!(reads.load(std::sync::atomic::Ordering::Relaxed), 1);
        assert_eq!(errors.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn native_change_trigger_is_initialized_only_when_polled() {
        let reader = || {
            BoxService::new(service_fn(|()| async {
                Ok::<_, BoxError>(SystemProxyConfig::default())
            }))
        };
        let trigger_initialized = |layer: &SystemProxyLayer| match &layer.refresh.trigger {
            Some(SystemProxyConfigChangeTrigger::Platform(trigger)) => trigger.is_initialized(),
            _ => panic!("expected the default platform change trigger"),
        };

        let disabled = SystemProxyLayer::from_cached_with_reader(
            SystemProxyConfig::default(),
            Duration::ZERO,
            reader(),
        )
        .with_config_refresh(false);
        assert!(!trigger_initialized(&disabled));
        disabled.config().await.unwrap();
        assert!(!trigger_initialized(&disabled));

        let enabled = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader());
        assert!(!trigger_initialized(&enabled));
        enabled.config().await.unwrap();
        assert!(trigger_initialized(&enabled));
    }

    #[tokio::test]
    async fn concurrent_cold_failure_is_shared_without_a_retry_convoy() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let read_started = Arc::new(tokio::sync::Notify::new());
        let release_read = Arc::new(tokio::sync::Notify::new());
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            let read_started = read_started.clone();
            let release_read = release_read.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let read_started = read_started.clone();
                let release_read = release_read.clone();
                async move {
                    if call == 0 {
                        read_started.notify_one();
                        release_read.notified().await;
                        Err(std::io::Error::other("temporary platform read failure").into())
                    } else {
                        Ok(SystemProxyConfig::default())
                    }
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);

        let release = async {
            read_started.notified().await;
            release_read.notify_one();
        };
        let (first, second, third, ()) = tokio::time::timeout(Duration::from_secs(5), async {
            tokio::join!(layer.config(), layer.config(), layer.config(), release,)
        })
        .await
        .expect("concurrent cold configuration load should complete");

        first.unwrap_err();
        second.unwrap_err();
        third.unwrap_err();
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);

        layer.config().await.unwrap();
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn cancelled_cold_load_releases_single_flight_lock() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    if call == 0 {
                        std::future::pending::<()>().await;
                    }
                    Ok::<_, BoxError>(SystemProxyConfig::default())
                }
            }
        }));
        let layer = SystemProxyLayer::new_with_reader(Duration::from_mins(1), reader);

        tokio::time::timeout(Duration::from_millis(10), layer.config())
            .await
            .unwrap_err();
        assert!(layer.cached_config().is_none());
        layer.config().await.unwrap();
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 2);
    }

    #[tokio::test]
    async fn cancelled_stale_refresh_releases_single_flight_lock() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    if call == 1 {
                        std::future::pending::<()>().await;
                    }
                    let host = if call == 0 { "old.proxy" } else { "new.proxy" };
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        host,
                        8080,
                    )))
                }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::ZERO, reader)
            .await
            .unwrap();

        tokio::time::timeout(Duration::from_millis(10), layer.config())
            .await
            .unwrap_err();
        let fresh = layer.config().await.unwrap();

        assert_eq!(
            fresh.http_proxy().unwrap().address.host.to_str(),
            "new.proxy"
        );
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn stale_refresh_is_single_flight_and_concurrent_calls_use_stale_config() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let refresh_started = Arc::new(tokio::sync::Notify::new());
        let allow_refresh = Arc::new(tokio::sync::Notify::new());
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            let refresh_started = refresh_started.clone();
            let allow_refresh = allow_refresh.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                let refresh_started = refresh_started.clone();
                let allow_refresh = allow_refresh.clone();
                async move {
                    if call > 0 {
                        refresh_started.notify_one();
                        allow_refresh.notified().await;
                    }
                    let host = if call == 0 { "old.proxy" } else { "new.proxy" };
                    Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                        Protocol::HTTP,
                        host,
                        8080,
                    )))
                }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::ZERO, reader)
            .await
            .unwrap();

        let refresh_layer = layer.clone();
        let refresh = tokio::spawn(async move { refresh_layer.config().await });
        tokio::time::timeout(Duration::from_secs(5), refresh_started.notified())
            .await
            .expect("stale configuration refresh should start");

        let stale = tokio::time::timeout(Duration::from_millis(100), layer.config())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            stale.http_proxy().unwrap().address.host.to_str(),
            "old.proxy"
        );
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 2);

        allow_refresh.notify_one();
        let fresh = tokio::time::timeout(Duration::from_secs(5), refresh)
            .await
            .expect("stale configuration refresh should complete")
            .unwrap()
            .unwrap();
        assert_eq!(
            fresh.http_proxy().unwrap().address.host.to_str(),
            "new.proxy"
        );
    }

    #[tokio::test]
    async fn failed_system_config_refresh_retains_snapshot_and_remains_retryable() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    match call {
                        0 => Ok(SystemProxyConfig::default().with_http_proxy(proxy(
                            Protocol::HTTP,
                            "old.proxy",
                            8080,
                        ))),
                        1 => Err(std::io::Error::other("temporary platform read failure").into()),
                        _ => Ok(SystemProxyConfig::default().with_http_proxy(proxy(
                            Protocol::HTTP,
                            "new.proxy",
                            8080,
                        ))),
                    }
                }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::ZERO, reader)
            .await
            .unwrap();

        let stale = layer.config().await.unwrap();
        assert_eq!(
            stale.http_proxy().unwrap().address.host.to_str(),
            "old.proxy"
        );
        let fresh = layer.config().await.unwrap();
        assert_eq!(
            fresh.http_proxy().unwrap().address.host.to_str(),
            "new.proxy"
        );
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn a_pac_uri_discovered_by_refresh_is_used_by_the_existing_service() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                let call = calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    if call == 0 {
                        Ok::<_, BoxError>(SystemProxyConfig::default().with_http_proxy(proxy(
                            Protocol::HTTP,
                            "fixed.proxy",
                            8080,
                        )))
                    } else {
                        Ok(SystemProxyConfig::default()
                            .with_pac_uri("https://config.example/proxy.pac".parse().unwrap()))
                    }
                }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::from_mins(1), reader)
            .await
            .unwrap();
        let factory_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let factory = service_fn({
            let factory_calls = factory_calls.clone();
            move |_uri: Uri| {
                factory_calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    Ok::<_, Infallible>(service_fn(|_request| async move {
                        Ok::<_, Infallible>(Some(ProxyRoutes::from(proxy(
                            Protocol::HTTP,
                            "pac.proxy",
                            8080,
                        ))))
                    }))
                }
            }
        });
        let (inner, seen) = recorder();
        let service = layer.clone().with_pac_service(factory).into_layer(inner);

        service
            .serve(TestInput::new("http://example.com/first"))
            .await
            .unwrap();
        layer.config.refresh_after_nanos.store(0, Ordering::Release);
        service
            .serve(TestInput::new("http://example.com/second"))
            .await
            .unwrap();

        let seen = seen.lock();
        let hosts = seen
            .iter()
            .map(|routes| {
                routes.as_ref().unwrap().as_slice()[0]
                    .proxy_address()
                    .unwrap()
                    .address
                    .host
                    .to_str()
                    .into_owned()
            })
            .collect::<Vec<_>>();
        assert_eq!(hosts, ["fixed.proxy", "pac.proxy"]);
        assert_eq!(factory_calls.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn system_config_cache_honors_a_custom_ttl() {
        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let reader = BoxService::new(service_fn({
            let calls = calls.clone();
            move |()| {
                calls.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async { Ok::<_, BoxError>(SystemProxyConfig::default()) }
            }
        }));
        let layer = SystemProxyLayer::try_from_system_with_reader(Duration::from_mins(1), reader)
            .await
            .unwrap();

        for _ in 0..10 {
            drop(layer.config().await.unwrap());
        }
        assert_eq!(calls.load(std::sync::atomic::Ordering::Relaxed), 1);
    }
}