praxis-proxy-core 0.7.4

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 Praxis Contributors

//! URL-aware target preparation.

use std::net::{IpAddr, SocketAddr};

use pingora_core::upstreams::peer::HttpPeer;

use super::peer::{AddressResolutionError, resolve_host_cached, resolve_host_per_call};
use crate::{connectivity::normalize_mapped_ipv4, subrequest::SubRequest};

/// Failure categories for URL target preparation.
///
/// `Display` for every variant is credential-safe: it never echoes the input
/// URL or any userinfo/query component. `PolicyRejected` keeps its `source` for
/// programmatic access but does not format caller-supplied data.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum UrlTargetError {
    /// The target could not be parsed or is structurally disallowed.
    #[error(transparent)]
    InvalidTarget(#[from] InvalidTarget),
    /// DNS resolution failed or returned no usable address.
    #[error(transparent)]
    Resolve(#[from] AddressResolutionError),
    /// The caller's validation hook rejected the resolved address set.
    #[error("address policy rejected the resolved target")]
    PolicyRejected(#[source] Box<dyn std::error::Error + Send + Sync>),
    /// The absolute deadline elapsed during preparation.
    #[error("target preparation deadline exceeded")]
    DeadlineExceeded,
}

/// Why a target was rejected at parse time. Carries a parser diagnostic or a
/// sanitized component only — never the original URL.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum InvalidTarget {
    /// Parser diagnostic (NOT the input URL).
    #[error("malformed target: {0}")]
    Malformed(String),
    /// The scheme token only (e.g. `ftp`).
    #[error("unsupported scheme: {0}")]
    UnsupportedScheme(String),
    /// Absent OR present-but-empty/bracket-only host.
    #[error("target has no host")]
    MissingHost,
    /// Any userinfo (including empty `@host`). No value stored.
    #[error("target must not contain userinfo")]
    UserinfoPresent,
    /// A fragment was present. No value stored.
    #[error("target must not contain a fragment")]
    FragmentPresent,
    /// The offending port token only.
    #[error("invalid port: {0}")]
    InvalidPort(String),
    /// Malformed IP-literal token only (e.g. `gggg::1`).
    #[error("invalid host literal: {0}")]
    InvalidHost(String),
}

/// DNS resolution choice for an absolute URL target.
///
/// Select [`OperatorCached`](Self::OperatorCached) for an operator-configured
/// destination. Select [`ClientPerCall`](Self::ClientPerCall) for a destination
/// selected by a client; this performs a fresh lookup without touching the
/// shared positive/negative cache or joining its in-flight lookups. IP literals
/// skip DNS under either policy. The caller must still validate every address.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum UrlResolutionPolicy {
    /// Reuse the bounded process-wide DNS cache for configured destinations.
    OperatorCached,
    /// Resolve a client-selected hostname for this call only.
    ///
    /// # Shared-pool limitation
    ///
    /// Every `ClientPerCall` lookup competes for one process-wide admission pool
    /// (see `MAX_PER_CALL_DNS_LOOKUPS` in [`super::peer`]). A slot is held until
    /// the blocking `getaddrinfo` returns, even after the calling request's
    /// deadline has elapsed and it has stopped waiting. A client aiming many
    /// calls at a nameserver that never answers can therefore pin every slot and
    /// stall other per-call lookups until those blocking resolutions drain.
    /// Releasing a slot before its lookup finishes would uncap the number of
    /// in-flight blocking resolutions, which is the worse failure, so the bound
    /// is deliberate. Operators who expose this policy to untrusted callers
    /// should add caller-side rate limiting and watch per-call DNS saturation;
    /// a stricter guarantee needs a resolver with a bounded per-lookup time.
    ClientPerCall,
}

/// Resolve a bare host (no port, no IPv6 brackets) to its RAW address answers,
/// BEFORE normalization/dedup. `SystemResolver` returns the complete cached
/// set; test doubles return scripted raw answers. Never returns `Ok(vec![])`.
pub(crate) trait HostResolver {
    /// Resolve `host` (already stripped of port and IPv6 brackets) to its raw
    /// address answers, before normalization/dedup. Never returns `Ok(vec![])`.
    fn resolve_host(&self, host: &str) -> impl Future<Output = Result<Vec<IpAddr>, AddressResolutionError>> + Send;
}

/// The production resolver: the real cache + detached-owner single-flight.
pub(crate) struct SystemResolver;

impl HostResolver for SystemResolver {
    async fn resolve_host(&self, host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
        resolve_host_cached(host).await
    }
}

/// Production resolver that bypasses the shared DNS cache.
struct PerCallResolver;

impl HostResolver for PerCallResolver {
    async fn resolve_host(&self, host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
        resolve_host_per_call(host).await
    }
}

/// Selects one of two resolvers without changing the preparation pipeline.
struct PolicyResolver<'resolver, C: HostResolver + Sync, P: HostResolver + Sync> {
    /// Which resolver to call for this URL.
    policy: UrlResolutionPolicy,
    /// Resolver for operator-configured hosts.
    cached: &'resolver C,
    /// Resolver for client-selected hosts.
    per_call: &'resolver P,
}

impl<C: HostResolver + Sync, P: HostResolver + Sync> HostResolver for PolicyResolver<'_, C, P> {
    async fn resolve_host(&self, host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
        match self.policy {
            UrlResolutionPolicy::OperatorCached => self.cached.resolve_host(host).await,
            UrlResolutionPolicy::ClientPerCall => self.per_call.resolve_host(host).await,
        }
    }
}

/// The validated shape of an absolute http(s) URL.
pub(crate) struct ParsedTarget {
    /// `true` for `https`, `false` for `http`.
    pub(crate) is_tls: bool,
    /// The `Host` header value: authority verbatim (host + explicit port,
    /// IPv6 brackets kept, no userinfo).
    pub(crate) host_authority: http::HeaderValue,
    /// Bracket-stripped bare host or IP, used for resolution AND SNI. Never empty.
    pub(crate) resolution_host: String,
    /// Explicit port, or the scheme default (80/443).
    pub(crate) effective_port: u16,
    /// Origin-form request target (path+query, or `/`).
    pub(crate) origin_form: http::Uri,
    /// `Some(ip)` when the host is a bare IP literal (DNS is skipped).
    pub(crate) literal_ip: Option<IpAddr>,
}

impl std::fmt::Debug for ParsedTarget {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ParsedTarget")
            .field("is_tls", &self.is_tls)
            .field("host_authority", &self.host_authority)
            .field("resolution_host", &self.resolution_host)
            .field("effective_port", &self.effective_port)
            // path only — the query is credential-sensitive (spec §4.5)
            .field("path", &self.origin_form.path())
            .field("literal_ip", &self.literal_ip)
            .finish()
    }
}

/// Parse and validate the shape of an absolute http(s) URL.
///
/// `http::Uri` accepts malformed hosts (`https://[]/`, `https://:443/`,
/// `[gggg::1]`) and invalid ports (`:99999`, `:abc`, `:0`), so this performs
/// targeted post-parse validation of the already-parsed authority — it does not
/// re-parse the raw URL and does not reuse `credentials::split_host_port`.
#[expect(clippy::too_many_lines, reason = "comprehensive validation logic from spec §4.3")]
#[expect(
    clippy::map_err_ignore,
    reason = "error is intentionally discarded; we only care that parsing failed"
)]
pub(crate) fn parse_target(url: &str) -> Result<ParsedTarget, InvalidTarget> {
    // Fragment: scan the raw input, independent of `http::Uri` fragment handling.
    if url.contains('#') {
        return Err(InvalidTarget::FragmentPresent);
    }

    let uri: http::Uri = url
        .parse()
        .map_err(|parse_err: http::uri::InvalidUri| InvalidTarget::Malformed(parse_err.to_string()))?;

    let is_tls = match uri.scheme_str() {
        Some("http") => false,
        Some("https") => true,
        Some(other) => return Err(InvalidTarget::UnsupportedScheme(other.to_owned())),
        None => return Err(InvalidTarget::Malformed("missing scheme".to_owned())),
    };

    let authority = uri.authority().ok_or(InvalidTarget::MissingHost)?;

    // Any `@` in the authority means userinfo is present (incl. empty `@host`).
    if authority.as_str().contains('@') {
        return Err(InvalidTarget::UserinfoPresent);
    }

    // Host, brackets kept (e.g. `[::1]`, `[]`, `""`, `example.com`).
    let host = authority.host();
    let (bracketed, bare) = match host.strip_prefix('[') {
        Some(rest) => match rest.strip_suffix(']') {
            Some(inner) => (true, inner),
            None => return Err(InvalidTarget::Malformed("unterminated IPv6 literal".to_owned())),
        },
        None => (false, host),
    };
    if bare.is_empty() {
        return Err(InvalidTarget::MissingHost);
    }

    // Literal-IP detection + malformed-literal rejection.
    let literal_ip = if bracketed {
        let ip = bare
            .parse::<IpAddr>()
            .map_err(|_| InvalidTarget::InvalidHost(bare.to_owned()))?;
        // Brackets denote an IPv6 literal (RFC 3986); a bracketed IPv4 would
        // produce the malformed Host `[a.b.c.d]`.
        let IpAddr::V6(v6) = ip else {
            return Err(InvalidTarget::InvalidHost(bare.to_owned()));
        };
        if v6.to_ipv4_mapped().is_some() {
            // Would normalize to bare IPv4 while SNI/Host keep the mapped form.
            return Err(InvalidTarget::InvalidHost(bare.to_owned()));
        }
        Some(IpAddr::V6(v6))
    } else {
        bare.parse::<IpAddr>().ok()
    };

    // Port token: the authority (userinfo already rejected) is `host[:port]`.
    let effective_port = {
        let remainder = authority.as_str().get(host.len()..).unwrap_or("");
        match remainder.strip_prefix(':') {
            None => {
                if is_tls {
                    443
                } else {
                    80
                }
            },
            Some(token) => match token.parse::<u16>() {
                Ok(port) if port >= 1 => port,
                _ => return Err(InvalidTarget::InvalidPort(token.to_owned())),
            },
        }
    };

    let host_authority = http::HeaderValue::from_str(authority.as_str())
        .map_err(|header_err| InvalidTarget::Malformed(header_err.to_string()))?;

    // `uri.path()` normalizes an empty authority-form path to `/` and always
    // carries a leading slash; `path_and_query().as_str()` does not (a
    // query-only URL yields `?q=x`, which is not a standalone origin-form URI).
    let origin_form: http::Uri = match uri.query() {
        Some(query) => format!("{}?{query}", uri.path()),
        None => uri.path().to_owned(),
    }
    .parse()
    .map_err(|parse_err: http::uri::InvalidUri| InvalidTarget::Malformed(parse_err.to_string()))?;

    Ok(ParsedTarget {
        is_tls,
        host_authority,
        resolution_host: bare.to_owned(),
        effective_port,
        origin_form,
        literal_ip,
    })
}

/// Validate an absolute HTTP(S) target without resolving or connecting to it.
///
/// Use this while loading static configuration. Runtime callers should use
/// [`prepare_url_target`] to validate resolved addresses and construct peers.
///
/// # Errors
///
/// Returns [`InvalidTarget`] when the URL is malformed or is not an absolute
/// HTTP(S) target permitted by the shared target rules.
pub fn validate_url_target(url: &str) -> Result<(), InvalidTarget> {
    parse_target(url).map(|_target| ())
}

/// A frozen, validated dial target. All addresses have already passed the
/// caller's validation hook. Peers are reachable only AFTER the request is
/// authority-bound via [`PreparedTarget::bind`], which makes the authority
/// invariant structural rather than a convention the caller must remember.
///
/// ```compile_fail
/// # async fn f(t: praxis_core::connectivity::PreparedTarget) {
/// // `PreparedTarget` exposes no peers: dialing requires `bind` first.
/// let _ = t.peers();
/// # }
/// ```
pub struct PreparedTarget {
    /// Whether the target uses TLS (`https`).
    is_tls: bool,
    /// The authority that will be bound as `Host`.
    host_authority: http::HeaderValue,
    /// The SNI string for TLS peers (empty for `http`).
    sni: String,
    /// The origin-form request target (path+query).
    origin_form: http::Uri,
    /// Validated dial addresses, in resolver order.
    addresses: Vec<SocketAddr>,
}

impl std::fmt::Debug for PreparedTarget {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PreparedTarget")
            .field("is_tls", &self.is_tls)
            .field("host_authority", &self.host_authority)
            .field("sni", &self.sni)
            // path only — the query is credential-sensitive (spec §4.5)
            .field("path", &self.origin_form.path())
            .field("addresses", &self.addresses)
            .finish()
    }
}

impl PreparedTarget {
    /// Freeze a validated target. Crate-private: only the preparation pipeline
    /// constructs one, after the validation hook has accepted `addresses`.
    pub(crate) fn new(
        is_tls: bool,
        host_authority: http::HeaderValue,
        sni: String,
        origin_form: http::Uri,
        addresses: Vec<SocketAddr>,
    ) -> Self {
        Self {
            is_tls,
            host_authority,
            // SNI is meaningful only for TLS; keep it empty for http so `sni()`
            // and `peer_for` agree.
            sni: if is_tls { sni } else { String::new() },
            origin_form,
            addresses,
        }
    }

    /// Consume the target and a request, OVERWRITE the request's `Host` with the
    /// URL authority and its target with the origin-form path+query, and return
    /// a [`PreparedSubrequest`] from which peers can be dialed. Infallible — the
    /// authority `HeaderValue` and origin-form `Uri` were built during parsing.
    #[must_use]
    pub fn bind(self, mut request: SubRequest) -> PreparedSubrequest {
        request.headers.insert(http::header::HOST, self.host_authority);
        request.uri = self.origin_form;
        PreparedSubrequest {
            is_tls: self.is_tls,
            sni: self.sni,
            addresses: self.addresses,
            request,
        }
    }

    /// The validated addresses, in resolver order. Inspection/logging only.
    #[must_use]
    pub fn addresses(&self) -> &[SocketAddr] {
        &self.addresses
    }

    /// The URL authority that will be bound as `Host`.
    #[must_use]
    pub fn host_authority(&self) -> &http::HeaderValue {
        &self.host_authority
    }

    /// Whether the target is `https`.
    #[must_use]
    pub fn is_tls(&self) -> bool {
        self.is_tls
    }

    /// The SNI string used for `https` peers (the URL host; empty for `http`).
    #[must_use]
    pub fn sni(&self) -> &str {
        &self.sni
    }
}

/// A validated target whose request has been authority-bound. The request is
/// exposed only by shared reference, so `Host` cannot diverge from the URL
/// authority or be mutated between fallback attempts.
///
/// ```
/// use std::time::{Duration, Instant};
///
/// use praxis_core::{connectivity::prepare_url_target, subrequest::SubRequest};
///
/// tokio::runtime::Runtime::new()
///     .expect("runtime")
///     .block_on(async {
///         let target = prepare_url_target(
///             "http://127.0.0.1:9/",
///             Instant::now() + Duration::from_secs(5),
///             |_| Ok(()),
///         )
///         .await
///         .expect("prepare");
///         let prepared = target.bind(SubRequest {
///             method: http::Method::GET,
///             uri: "/".parse().unwrap(),
///             headers: http::HeaderMap::new(),
///             body: bytes::Bytes::new(),
///         });
///         assert_eq!(prepared.peers().count(), 1); // peers reachable only after bind
///     });
/// ```
pub struct PreparedSubrequest {
    /// Whether the target uses TLS.
    is_tls: bool,
    /// The SNI string for TLS peers (empty for `http`).
    sni: String,
    /// Validated dial addresses, in resolver order.
    addresses: Vec<SocketAddr>,
    /// The authority-bound request.
    request: SubRequest,
}

impl PreparedSubrequest {
    /// Peers for every validated address, in resolver order. Primary path.
    pub fn peers(&self) -> impl Iterator<Item = HttpPeer> + '_ {
        self.addresses
            .iter()
            .map(|addr| peer_for(*addr, self.is_tls, &self.sni))
    }

    /// Peer for the address at `index`, or `None` if out of range. Indexed
    /// fallback for callers that track attempt position.
    #[must_use]
    pub fn peer_at(&self, index: usize) -> Option<HttpPeer> {
        self.addresses
            .get(index)
            .map(|addr| peer_for(*addr, self.is_tls, &self.sni))
    }

    /// The validated addresses, in resolver order. Inspection/logging only.
    #[must_use]
    pub fn addresses(&self) -> &[SocketAddr] {
        &self.addresses
    }

    /// The authority-bound request, by shared reference only.
    #[must_use]
    pub fn request(&self) -> &SubRequest {
        &self.request
    }
}

/// Build a peer for one validated address.
///
/// Fail-closed: a TLS peer with an empty SNI would switch pingora to
/// `VerificationMode::SkipAll`, so this refuses to build one. Combined with the
/// parse-time empty-host rejection, an empty TLS SNI is unreachable. This is a
/// real runtime check (present in release), not a `debug_assert!`.
fn peer_for(addr: SocketAddr, is_tls: bool, sni: &str) -> HttpPeer {
    if is_tls {
        assert!(
            !sni.is_empty(),
            "BUG: refusing to build a TLS peer with an empty SNI (non-empty SNI invariant violated)"
        );
        HttpPeer::new(addr, true, sni.to_owned())
    } else {
        HttpPeer::new(addr, false, String::new())
    }
}

/// Parse an absolute http(s) URL, resolve it to the complete address set,
/// validate that set atomically, and freeze a target that can be dialed (with
/// fallback) without re-resolving DNS.
///
/// The parsed URL authority is authoritative: [`PreparedTarget::bind`] later
/// overwrites the request's `Host` with it and sets the SNI from it, so the
/// resolved address, TLS identity, request target, and HTTP authority cannot
/// diverge (the SSRF / DNS-rebinding / TLS-identity-confusion surface).
///
/// `deadline` is the caller's existing absolute deadline; it bounds the whole
/// preparation, including waiting for the inflight resolution result.
///
/// `validate` is called AT MOST ONCE, and only after successful parsing and
/// resolution, on the complete normalized+deduplicated address set. Invalid
/// URLs, resolution failures, and any deadline expiry detected on entry,
/// post-parse, or pre-validate call it ZERO times. Because the synchronous hook
/// cannot be preempted, a hook that itself overruns the deadline still runs to
/// completion and its side effects still occur; the call then returns
/// [`UrlTargetError::DeadlineExceeded`] instead of a `PreparedTarget`. So
/// `Err(DeadlineExceeded)` does NOT by itself guarantee the hook was not
/// invoked — a caller whose hook has observable side effects must treat them as
/// possibly-executed on `DeadlineExceeded`. Returning `Err` from the hook
/// prevents every connection attempt.
///
/// Deciding whether private/loopback addresses are allowed is the hook's job,
/// not this function's; IP-literal targets still pass through the hook.
/// This compatibility entry point uses the shared cached resolver. New callers
/// choosing between configured and client-selected URLs should use
/// [`prepare_url_target_with_policy`].
///
/// # Errors
///
/// Returns [`UrlTargetError`]. Every variant's `Display` is credential-safe: it
/// never echoes the input URL, userinfo, or query.
///
/// ```
/// use std::time::{Duration, Instant};
///
/// use praxis_core::connectivity::prepare_url_target;
///
/// tokio::runtime::Runtime::new()
///     .expect("runtime")
///     .block_on(async {
///         // Port 9 (discard) is never dialed — preparation does not connect.
///         let target = prepare_url_target(
///             "http://127.0.0.1:9/health",
///             Instant::now() + Duration::from_secs(5),
///             |_addrs| Ok(()), // the caller's SSRF policy lives here
///         )
///         .await
///         .expect("a literal target prepares without DNS");
///         assert!(!target.is_tls());
///         assert_eq!(target.host_authority().to_str().unwrap(), "127.0.0.1:9");
///     });
/// ```
pub async fn prepare_url_target<F>(
    url: &str,
    deadline: std::time::Instant,
    validate: F,
) -> Result<PreparedTarget, UrlTargetError>
where
    F: FnOnce(&[SocketAddr]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
{
    prepare_url_target_with_policy(url, deadline, validate, UrlResolutionPolicy::OperatorCached).await
}

/// Prepare and validate a URL using an explicit DNS resolution policy.
///
/// This has the same frozen-address and authority-binding contract as
/// [`prepare_url_target`]. Both modes validate the complete resolved set before
/// any dial. This function only prepares a target; callers may bind it for
/// buffered or streaming execution. Use [`UrlResolutionPolicy::ClientPerCall`]
/// for client-selected destinations to keep them out of the shared DNS cache.
///
/// # Errors
///
/// Returns [`UrlTargetError`] for an invalid URL, DNS or policy rejection, or
/// deadline expiry. Its display text does not include the input URL.
///
/// ```
/// use std::time::{Duration, Instant};
///
/// use praxis_core::{
///     connectivity::{UrlResolutionPolicy, prepare_url_target_with_policy},
///     subrequest::SubRequest,
/// };
///
/// tokio::runtime::Runtime::new()
///     .expect("runtime")
///     .block_on(async {
///         // An IP literal skips DNS, so `ClientPerCall` prepares offline.
///         let target = prepare_url_target_with_policy(
///             "https://[::1]:8443/v1/models",
///             Instant::now() + Duration::from_secs(5),
///             |_addrs| Ok(()), // the caller's SSRF policy lives here
///             UrlResolutionPolicy::ClientPerCall,
///         )
///         .await
///         .expect("a literal target prepares without DNS");
///         assert!(target.is_tls());
///         assert_eq!(target.host_authority().to_str().unwrap(), "[::1]:8443");
///         assert_eq!(target.sni(), "::1"); // SNI drops the brackets
///
///         // Binding overwrites the request's Host with the URL authority.
///         let prepared = target.bind(SubRequest {
///             method: http::Method::GET,
///             uri: "/ignored".parse().unwrap(),
///             headers: http::HeaderMap::new(),
///             body: bytes::Bytes::new(),
///         });
///         assert_eq!(prepared.peers().count(), 1);
///         assert_eq!(
///             prepared
///                 .request()
///                 .headers
///                 .get(http::header::HOST)
///                 .unwrap()
///                 .to_str()
///                 .unwrap(),
///             "[::1]:8443",
///         );
///     });
/// ```
pub async fn prepare_url_target_with_policy<F>(
    url: &str,
    deadline: std::time::Instant,
    validate: F,
    policy: UrlResolutionPolicy,
) -> Result<PreparedTarget, UrlTargetError>
where
    F: FnOnce(&[SocketAddr]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
{
    let cached = SystemResolver;
    let per_call = PerCallResolver;
    let resolver = PolicyResolver {
        policy,
        cached: &cached,
        per_call: &per_call,
    };
    prepare_url_target_with_resolver(url, deadline, validate, &resolver).await
}

/// The generic preparation pipeline. The public policy-aware wrapper binds a
/// production resolver; tests inject a `FakeResolver`.
#[expect(
    clippy::too_many_lines,
    reason = "linear pin-before-dial pipeline with four deadline checkpoints; splitting would break the one-clock invariant"
)]
pub(crate) async fn prepare_url_target_with_resolver<R: HostResolver + Sync>(
    url: &str,
    deadline: std::time::Instant,
    validate: impl FnOnce(&[SocketAddr]) -> Result<(), Box<dyn std::error::Error + Send + Sync>> + Send,
    resolver: &R,
) -> Result<PreparedTarget, UrlTargetError> {
    // One clock for every checkpoint AND timeout_at, so `tokio::time::pause()`
    // drives all of them uniformly.
    let deadline = tokio::time::Instant::from_std(deadline);

    // Checkpoint 1: on entry.
    if tokio::time::Instant::now() >= deadline {
        return Err(UrlTargetError::DeadlineExceeded);
    }

    let parsed = parse_target(url)?;

    // Checkpoint 2: after parsing / shape validation.
    if tokio::time::Instant::now() >= deadline {
        return Err(UrlTargetError::DeadlineExceeded);
    }

    // Resolve (DNS hosts) or skip (IP literals). The resolver never returns an
    // empty set (zero answers → Resolve).
    let raw_ips: Vec<IpAddr> = match parsed.literal_ip {
        Some(ip) => vec![ip],
        None => tokio::time::timeout_at(deadline, resolver.resolve_host(&parsed.resolution_host))
            .await
            .map_err(|_elapsed| UrlTargetError::DeadlineExceeded)??,
    };

    // Attach the effective port, normalize IPv4-mapped IPv6, order-preserving dedup.
    let mut seen = std::collections::HashSet::new();
    let addresses: Vec<SocketAddr> = raw_ips
        .into_iter()
        .map(normalize_mapped_ipv4)
        .map(|ip| SocketAddr::new(ip, parsed.effective_port))
        .filter(|sa| seen.insert(*sa))
        .collect();

    if addresses.is_empty() {
        return Err(UrlTargetError::Resolve(AddressResolutionError::Empty(
            parsed.resolution_host,
        )));
    }

    // Checkpoint 3: immediately before validate.
    if tokio::time::Instant::now() >= deadline {
        return Err(UrlTargetError::DeadlineExceeded);
    }

    validate(&addresses).map_err(UrlTargetError::PolicyRejected)?;

    // Checkpoint 4: after a successful validate, before freezing.
    if tokio::time::Instant::now() >= deadline {
        return Err(UrlTargetError::DeadlineExceeded);
    }

    Ok(PreparedTarget::new(
        parsed.is_tls,
        parsed.host_authority,
        parsed.resolution_host,
        parsed.origin_form,
        addresses,
    ))
}

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::assertions_on_result_states,
    clippy::min_ident_chars,
    clippy::shadow_unrelated,
    clippy::arithmetic_side_effects,
    reason = "tests"
)]
mod tests {
    use std::time::{Duration, Instant};

    use super::*;
    use crate::subrequest::{
        StreamLimits, SubRequestClient, SubRequestConnector, SubRequestConnectorOptions, SubRequestError,
        UrlSubRequestError,
    };

    #[test]
    fn invalid_target_display_is_token_only() {
        assert_eq!(
            InvalidTarget::UnsupportedScheme("ftp".to_owned()).to_string(),
            "unsupported scheme: ftp"
        );
        assert_eq!(
            InvalidTarget::InvalidPort("99999".to_owned()).to_string(),
            "invalid port: 99999"
        );
        assert!(!InvalidTarget::UserinfoPresent.to_string().contains('@'));
    }

    #[test]
    fn policy_rejected_display_hides_the_source() {
        let source: Box<dyn std::error::Error + Send + Sync> = "secret-token-abc".into();
        let err = UrlTargetError::PolicyRejected(source);
        assert!(
            !err.to_string().contains("secret-token-abc"),
            "PolicyRejected Display must not format caller-supplied data: {err}"
        );
    }

    #[test]
    fn invalid_target_converts_into_url_target_error() {
        let err: UrlTargetError = InvalidTarget::MissingHost.into();
        assert!(matches!(err, UrlTargetError::InvalidTarget(InvalidTarget::MissingHost)));
    }

    #[expect(clippy::panic, reason = "test utility panics to fail the test on parse error")]
    fn parse_ok(url: &str) -> ParsedTarget {
        parse_target(url).unwrap_or_else(|e| panic!("expected {url} to parse: {e}"))
    }

    #[test]
    fn parses_https_default_port_and_host_header() {
        let p = parse_ok("https://api.example.com/v1/models?a=1");
        assert!(p.is_tls);
        assert_eq!(p.effective_port, 443);
        assert_eq!(p.host_authority.to_str().unwrap(), "api.example.com");
        assert_eq!(p.resolution_host, "api.example.com");
        assert_eq!(p.origin_form.to_string(), "/v1/models?a=1");
        assert!(p.literal_ip.is_none());
    }

    #[test]
    fn parses_http_default_port() {
        let p = parse_ok("http://example.com/");
        assert!(!p.is_tls);
        assert_eq!(p.effective_port, 80);
        assert_eq!(p.origin_form.to_string(), "/");
    }

    #[test]
    fn explicit_non_default_port_flows_to_host_and_dial() {
        let p = parse_ok("https://example.com:8443/x");
        assert_eq!(p.effective_port, 8443);
        assert_eq!(p.host_authority.to_str().unwrap(), "example.com:8443");
        assert_eq!(p.resolution_host, "example.com");
    }

    #[test]
    fn parses_ipv4_literal_as_literal_ip() {
        let p = parse_ok("http://127.0.0.1/");
        assert_eq!(p.literal_ip, Some("127.0.0.1".parse::<IpAddr>().unwrap()));
        assert_eq!(p.effective_port, 80);
        assert_eq!(p.resolution_host, "127.0.0.1");
        assert_eq!(p.host_authority.to_str().unwrap(), "127.0.0.1");
    }

    #[test]
    fn parses_bracketed_ipv6_literal() {
        let p = parse_ok("https://[2001:db8::1]:8443/");
        assert_eq!(p.literal_ip, Some("2001:db8::1".parse::<IpAddr>().unwrap()));
        assert_eq!(p.resolution_host, "2001:db8::1");
        assert_eq!(p.host_authority.to_str().unwrap(), "[2001:db8::1]:8443");
        assert_eq!(p.effective_port, 8443);
    }

    #[test]
    fn empty_path_defaults_to_slash() {
        assert_eq!(parse_ok("http://h").origin_form.to_string(), "/");
    }

    #[test]
    fn query_only_url_gets_root_path() {
        let p = parse_ok("https://api.example.com?token=x");
        assert_eq!(
            p.origin_form.to_string(),
            "/?token=x",
            "an empty path with a query must default to /, not the leading-slash-less ?token=x"
        );
    }

    #[test]
    fn rejects_unsupported_scheme() {
        assert!(matches!(
            parse_target("ftp://h/"),
            Err(InvalidTarget::UnsupportedScheme(s)) if s == "ftp"
        ));
    }

    #[test]
    fn rejects_userinfo_including_empty() {
        assert!(matches!(
            parse_target("http://user:pw@h/"),
            Err(InvalidTarget::UserinfoPresent)
        ));
        assert!(matches!(
            parse_target("http://@h/"),
            Err(InvalidTarget::UserinfoPresent)
        ));
    }

    #[test]
    fn rejects_fragment() {
        assert!(matches!(
            parse_target("http://h/p#frag"),
            Err(InvalidTarget::FragmentPresent)
        ));
    }

    #[test]
    fn rejects_empty_and_bracket_only_hosts() {
        assert!(matches!(parse_target("https://[]/"), Err(InvalidTarget::MissingHost)));
        assert!(matches!(
            parse_target("https://[]:443/"),
            Err(InvalidTarget::MissingHost)
        ));
        assert!(matches!(parse_target("https://:443/"), Err(InvalidTarget::MissingHost)));
    }

    #[test]
    #[expect(clippy::panic, reason = "test panics on unexpected parse result")]
    fn rejects_malformed_ip_literals_with_token_only() {
        for (url, token) in [("https://[gggg::1]/", "gggg::1"), ("https://[vFF.abc]/", "vFF.abc")] {
            match parse_target(url) {
                Err(InvalidTarget::InvalidHost(t)) => {
                    assert_eq!(t, token);
                    assert!(!t.contains("https://"), "InvalidHost must carry the token only");
                },
                other => panic!("expected InvalidHost({token}), got {other:?}"),
            }
        }
    }

    #[test]
    fn rejects_ipv4_mapped_ipv6_literal() {
        assert!(matches!(
            parse_target("https://[::ffff:1.2.3.4]/"),
            Err(InvalidTarget::InvalidHost(_))
        ));
    }

    #[test]
    fn rejects_bracketed_ipv4_literal() {
        assert!(
            matches!(
                parse_target("https://[127.0.0.1]/"),
                Err(InvalidTarget::InvalidHost(t)) if t == "127.0.0.1"
            ),
            "brackets denote an IPv6 literal (RFC 3986); a bracketed IPv4 is the malformed Host [127.0.0.1]"
        );
    }

    #[test]
    #[expect(clippy::panic, reason = "test panics on unexpected parse result")]
    fn rejects_bad_ports_with_token_only() {
        for (url, token) in [
            ("http://h:99999/", "99999"),
            ("http://h:abc/", "abc"),
            ("http://h:-1/", "-1"),
            ("http://h:0/", "0"),
            ("http://h:8080evil/", "8080evil"),
            ("http://[::1]:abc/", "abc"),
            ("http://[::1]:/", ""),
        ] {
            match parse_target(url) {
                Err(InvalidTarget::InvalidPort(t)) => assert_eq!(t, token, "for {url}"),
                other => panic!("expected InvalidPort({token:?}) for {url}, got {other:?}"),
            }
        }
    }

    #[test]
    fn parse_errors_never_contain_the_input_url() {
        let url = "ftp://secret-user@host/path?token=abc";
        let err = parse_target(url).unwrap_err();
        assert!(!err.to_string().contains("secret-user"));
        assert!(!err.to_string().contains("token=abc"));

        let malformed = "https://exa mple.com/secret-path?q=leak";
        let merr = parse_target(malformed).unwrap_err();
        assert!(
            matches!(merr, InvalidTarget::Malformed(_)),
            "expected Malformed, got {merr:?}"
        );
        let disp = merr.to_string();
        assert!(!disp.contains("exa mple"), "Malformed Display leaked host: {disp}");
        assert!(!disp.contains("secret-path"), "Malformed Display leaked path: {disp}");
        assert!(!disp.contains("q=leak"), "Malformed Display leaked query: {disp}");
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    use pingora_core::upstreams::peer::{Peer as _, Scheme};

    fn target(is_tls: bool, authority: &str, sni: &str, addrs: &[&str]) -> PreparedTarget {
        PreparedTarget::new(
            is_tls,
            http::HeaderValue::from_str(authority).unwrap(),
            sni.to_owned(),
            "/p?q=1".parse().unwrap(),
            addrs.iter().map(|a| a.parse::<SocketAddr>().unwrap()).collect(),
        )
    }

    fn req_with_host(host: &str) -> SubRequest {
        let mut headers = http::HeaderMap::new();
        headers.insert(http::header::HOST, http::HeaderValue::from_str(host).unwrap());
        SubRequest {
            method: http::Method::GET,
            uri: "/original".parse().unwrap(),
            headers,
            body: bytes::Bytes::new(),
        }
    }

    #[test]
    fn accessors_expose_frozen_fields() {
        let t = target(true, "api.example.com:8443", "api.example.com", &["93.184.216.34:8443"]);
        assert!(t.is_tls());
        assert_eq!(t.sni(), "api.example.com");
        assert_eq!(t.host_authority().to_str().unwrap(), "api.example.com:8443");
        assert_eq!(t.addresses(), ["93.184.216.34:8443".parse::<SocketAddr>().unwrap()]);
    }

    #[test]
    fn http_target_has_empty_sni() {
        let t = target(false, "example.com", "example.com", &["93.184.216.34:80"]);
        assert_eq!(t.sni(), "", "sni() must be empty for http");
    }

    #[test]
    fn bind_overwrites_host_and_rewrites_target() {
        let t = target(true, "api.example.com:8443", "api.example.com", &["93.184.216.34:8443"]);
        let prepared = t.bind(req_with_host("attacker.example"));
        let req = prepared.request();
        assert_eq!(
            req.headers.get(http::header::HOST).unwrap().to_str().unwrap(),
            "api.example.com:8443",
            "bind must overwrite a conflicting caller Host"
        );
        assert_eq!(req.headers.get_all(http::header::HOST).iter().count(), 1);
        assert_eq!(
            req.uri.to_string(),
            "/p?q=1",
            "bind must rewrite the target to origin-form"
        );
    }

    #[test]
    fn peers_build_tls_peer_per_address_in_order() {
        let t = target(true, "h:8443", "h.example.com", &["10.0.0.1:8443", "10.0.0.2:8443"]);
        let prepared = t.bind(req_with_host("ignored"));
        let peers: Vec<_> = prepared.peers().collect();
        assert_eq!(peers.len(), 2);
        assert_eq!(peers[0].address().to_string(), "10.0.0.1:8443");
        assert_eq!(peers[1].address().to_string(), "10.0.0.2:8443");
        assert_eq!(peers[0].sni, "h.example.com");
        assert_eq!(peers[0].scheme, Scheme::HTTPS);
        assert_eq!(prepared.peer_at(1).unwrap().address().to_string(), "10.0.0.2:8443");
        assert!(prepared.peer_at(2).is_none());
    }

    #[test]
    fn http_peers_have_empty_sni_and_plain_scheme() {
        let t = target(false, "h", "", &["10.0.0.1:80"]);
        let prepared = t.bind(req_with_host("ignored"));
        let peer = prepared.peer_at(0).unwrap();
        assert_eq!(peer.sni, "");
        assert_eq!(peer.scheme, Scheme::HTTP);
    }

    #[test]
    #[should_panic(expected = "non-empty SNI")]
    fn peer_for_refuses_empty_tls_sni() {
        drop(peer_for("10.0.0.1:443".parse().unwrap(), true, ""));
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    use std::sync::{
        Arc as StdArc,
        atomic::{AtomicUsize, Ordering},
    };

    #[derive(Clone)]
    enum FakeDelay {
        None,
        /// Sleep on the real clock before returning (resolution-timeout test).
        Sleep(Duration),
        /// Advance the paused Tokio clock, then return immediately
        /// (pre-validate checkpoint test).
        Advance(Duration),
    }

    struct FakeResolver {
        calls: StdArc<AtomicUsize>,
        respond: Box<dyn Fn() -> Result<Vec<IpAddr>, AddressResolutionError> + Send + Sync>,
        delay: FakeDelay,
    }

    impl FakeResolver {
        fn ok(ips: Vec<IpAddr>) -> Self {
            Self {
                calls: StdArc::new(AtomicUsize::new(0)),
                respond: Box::new(move || Ok(ips.clone())),
                delay: FakeDelay::None,
            }
        }

        fn failing() -> Self {
            Self {
                calls: StdArc::new(AtomicUsize::new(0)),
                respond: Box::new(|| Err(AddressResolutionError::Empty("fake".to_owned()))),
                delay: FakeDelay::None,
            }
        }

        fn with_delay(mut self, delay: FakeDelay) -> Self {
            self.delay = delay;
            self
        }

        fn call_count(&self) -> usize {
            self.calls.load(Ordering::SeqCst)
        }
    }

    impl HostResolver for FakeResolver {
        async fn resolve_host(&self, _host: &str) -> Result<Vec<IpAddr>, AddressResolutionError> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            let out = (self.respond)();
            let delay = self.delay.clone();
            match delay {
                FakeDelay::None => {},
                FakeDelay::Sleep(d) => tokio::time::sleep(d).await,
                FakeDelay::Advance(d) => tokio::time::advance(d).await,
            }
            out
        }
    }

    #[tokio::test]
    async fn system_resolver_resolves_localhost() {
        let ips = SystemResolver
            .resolve_host("localhost")
            .await
            .expect("localhost resolves");
        assert!(ips.iter().all(IpAddr::is_loopback), "got {ips:?}");
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "checks both resolver policies with deterministic answers"
    )]
    async fn policy_aware_preparation_selects_injected_resolver() {
        let cached = FakeResolver::ok(vec!["203.0.113.7".parse().unwrap()]);
        let per_call = FakeResolver::ok(vec!["127.0.0.1".parse().unwrap()]);
        let cached_policy = PolicyResolver {
            policy: UrlResolutionPolicy::OperatorCached,
            cached: &cached,
            per_call: &per_call,
        };
        let target = prepare_url_target_with_resolver(
            "http://fixture.invalid:8123/",
            far_deadline(),
            |_| Ok(()),
            &cached_policy,
        )
        .await
        .expect("cached resolver prepares the target");
        assert_eq!(
            target.addresses()[0].ip(),
            "203.0.113.7".parse::<IpAddr>().unwrap(),
            "operator policy must select the cached resolver"
        );
        assert_eq!(cached.call_count(), 1, "cached resolver must run once");
        assert_eq!(per_call.call_count(), 0, "per-call resolver must remain unused");

        let per_call_policy = PolicyResolver {
            policy: UrlResolutionPolicy::ClientPerCall,
            cached: &cached,
            per_call: &per_call,
        };
        let target = prepare_url_target_with_resolver(
            "http://fixture.invalid:8123/",
            far_deadline(),
            |_| Ok(()),
            &per_call_policy,
        )
        .await
        .expect("per-call resolver prepares the target");
        assert_eq!(
            target.addresses()[0].ip(),
            "127.0.0.1".parse::<IpAddr>().unwrap(),
            "client policy must select the per-call resolver"
        );
        assert_eq!(cached.call_count(), 1, "per-call policy must not use cached resolver");
        assert_eq!(per_call.call_count(), 1, "per-call resolver must run once");
    }

    #[tokio::test]
    async fn policy_aware_preparation_wires_production_cache_bypass() {
        // Exercises the production `prepare_url_target_with_policy` wiring (real
        // SystemResolver / PerCallResolver), not an injected resolver. A reserved
        // `.invalid` host (RFC 6761: never resolvable) seeded with a TEST-NET-3
        // sentinel (RFC 5737: never returned by a real resolver) makes both arms
        // deterministic regardless of the machine's DNS behavior.
        let host = "production-wiring.praxis-test.invalid";
        let sentinel: IpAddr = "203.0.113.7".parse().unwrap();
        crate::connectivity::peer::seed_dns(host, &[sentinel]);
        let url = format!("http://{host}:9/");

        // OperatorCached reads the shared cache and returns the seeded answer.
        let cached =
            prepare_url_target_with_policy(&url, far_deadline(), |_| Ok(()), UrlResolutionPolicy::OperatorCached)
                .await
                .expect("cached policy reads the seeded answer");
        assert!(
            cached.addresses().iter().any(|addr| addr.ip() == sentinel),
            "OperatorCached must return the seeded cache answer: {:?}",
            cached.addresses()
        );

        // ClientPerCall bypasses the cache: it attempts a live lookup of an
        // unresolvable host instead of reading the seeded sentinel. Whether that
        // lookup fails (NXDOMAIN) or a hijacking resolver answers, it can never
        // return the sentinel, so its absence proves the bypass.
        let per_call =
            prepare_url_target_with_policy(&url, far_deadline(), |_| Ok(()), UrlResolutionPolicy::ClientPerCall).await;
        let leaked_sentinel = per_call
            .as_ref()
            .is_ok_and(|target| target.addresses().iter().any(|addr| addr.ip() == sentinel));
        assert!(
            !leaked_sentinel,
            "ClientPerCall must bypass the cache and never return the seeded sentinel: {per_call:?}"
        );
    }

    #[tokio::test]
    async fn fake_resolver_counts_calls() {
        let fake = FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]);
        let ips = fake.resolve_host("h").await.unwrap();
        assert_eq!(ips, vec!["1.2.3.4".parse::<IpAddr>().unwrap()]);
        assert_eq!(fake.call_count(), 1);
    }

    fn far_deadline() -> Instant {
        Instant::now() + Duration::from_secs(30)
    }

    #[tokio::test]
    async fn ip_literal_skips_dns_but_hits_hook() {
        for url in ["http://127.0.0.1/", "http://[::1]/"] {
            let fake = FakeResolver::ok(vec!["9.9.9.9".parse().unwrap()]);
            let seen: std::sync::Mutex<Vec<SocketAddr>> = std::sync::Mutex::new(Vec::new());
            let target = prepare_url_target_with_resolver(
                url,
                far_deadline(),
                |addrs| {
                    seen.lock().unwrap().extend_from_slice(addrs);
                    Ok(())
                },
                &fake,
            )
            .await
            .expect("literal resolves without DNS");
            assert_eq!(fake.call_count(), 0, "IP literals must not touch the resolver");
            assert_eq!(
                seen.lock().unwrap().len(),
                1,
                "the hook must see the literal address for {url}"
            );
            assert_eq!(target.addresses().len(), 1);
        }
    }

    #[tokio::test]
    async fn rejecting_hook_yields_policy_rejected_and_no_target() {
        let fake = FakeResolver::ok(vec![]);
        let err = prepare_url_target_with_resolver(
            "http://127.0.0.1/",
            far_deadline(),
            |_| Err("blocked by policy".into()),
            &fake,
        )
        .await
        .expect_err("a rejecting hook blocks the target");
        assert!(matches!(err, UrlTargetError::PolicyRejected(_)), "got {err}");
    }

    #[tokio::test]
    async fn dns_host_resolving_to_empty_set_yields_resolve_empty() {
        let fake = FakeResolver::ok(vec![]);
        let err = prepare_url_target_with_resolver("http://empty-set.test/", far_deadline(), |_| Ok(()), &fake)
            .await
            .expect_err("empty resolved set");
        assert!(
            matches!(err, UrlTargetError::Resolve(AddressResolutionError::Empty(_))),
            "a zero-address DNS answer must be a clean Resolve(Empty), not a peerless target: {err:?}"
        );
        assert_eq!(fake.call_count(), 1, "the DNS host must hit the resolver exactly once");
    }

    #[tokio::test]
    async fn complete_set_reaches_hook_normalized_and_deduped() {
        let fake = FakeResolver::ok(vec![
            "::ffff:1.2.3.4".parse().unwrap(),
            "1.2.3.4".parse().unwrap(),
            "9.9.9.9".parse().unwrap(),
        ]);
        let target = prepare_url_target_with_resolver("https://h.example.com/", far_deadline(), |_| Ok(()), &fake)
            .await
            .expect("resolves");
        assert_eq!(
            target.addresses(),
            [
                "1.2.3.4:443".parse::<SocketAddr>().unwrap(),
                "9.9.9.9:443".parse::<SocketAddr>().unwrap(),
            ],
            "mapped-IPv4 unwrapped, order-preserving dedup, effective port applied"
        );
    }

    #[tokio::test]
    async fn resolver_error_becomes_url_resolve_error_without_calling_hook() {
        let fake = FakeResolver::failing();
        let called = StdArc::new(std::sync::atomic::AtomicBool::new(false));
        let called_hook = StdArc::clone(&called);
        let err = prepare_url_target_with_resolver(
            "https://h.example.com/",
            far_deadline(),
            move |_| {
                called_hook.store(true, Ordering::SeqCst);
                Ok(())
            },
            &fake,
        )
        .await
        .expect_err("resolver failure surfaces");
        assert!(matches!(err, UrlTargetError::Resolve(_)), "got {err}");
        assert!(
            !called.load(Ordering::SeqCst),
            "the hook must not run on resolution failure"
        );
    }

    #[tokio::test]
    async fn no_re_resolve_across_fallback() {
        let fake = FakeResolver::ok(vec!["1.2.3.4".parse().unwrap(), "5.6.7.8".parse().unwrap()]);
        let target = prepare_url_target_with_resolver("http://h.example.com/", far_deadline(), |_| Ok(()), &fake)
            .await
            .expect("resolves");
        let prepared = target.bind(req_with_host("ignored"));
        let _peers: Vec<_> = prepared.peers().collect();
        assert_eq!(fake.call_count(), 1, "fallback must not re-resolve");
    }

    #[tokio::test]
    async fn deadline_expired_on_entry_for_literal() {
        let past = Instant::now();
        tokio::time::sleep(Duration::from_millis(10)).await;
        let fake = FakeResolver::ok(vec![]);
        let err = prepare_url_target_with_resolver("http://127.0.0.1/", past, |_| Ok(()), &fake)
            .await
            .expect_err("elapsed deadline");
        assert!(matches!(err, UrlTargetError::DeadlineExceeded));
    }

    #[tokio::test]
    async fn deadline_expires_during_resolution() {
        let fake =
            FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]).with_delay(FakeDelay::Sleep(Duration::from_secs(60)));
        let deadline = Instant::now() + Duration::from_millis(50);
        let err = prepare_url_target_with_resolver("https://h.example.com/", deadline, |_| Ok(()), &fake)
            .await
            .expect_err("resolution outruns the deadline");
        assert!(
            matches!(err, UrlTargetError::DeadlineExceeded),
            "timeout_at must wrap resolve_host: {err}"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn deadline_checkpoint_before_validate_fires() {
        let deadline = Instant::now() + Duration::from_secs(30);
        let fake =
            FakeResolver::ok(vec!["1.2.3.4".parse().unwrap()]).with_delay(FakeDelay::Advance(Duration::from_secs(31)));
        let called = StdArc::new(std::sync::atomic::AtomicBool::new(false));
        let called_hook = StdArc::clone(&called);
        let err = prepare_url_target_with_resolver(
            "https://h.example.com/",
            deadline,
            move |_| {
                called_hook.store(true, Ordering::SeqCst);
                Ok(())
            },
            &fake,
        )
        .await
        .expect_err("checkpoint 3 catches expiry");
        assert!(matches!(err, UrlTargetError::DeadlineExceeded));
        assert!(
            !called.load(Ordering::SeqCst),
            "pre-validate checkpoint must block the hook"
        );
    }

    #[tokio::test]
    #[allow(
        clippy::disallowed_methods,
        reason = "synchronous hook intentionally blocks the real clock to test checkpoint 4"
    )]
    async fn deadline_after_validate_returns_deadline_but_hook_ran() {
        let deadline = Instant::now() + Duration::from_millis(30);
        let fake = FakeResolver::ok(vec![]);
        let ran = StdArc::new(std::sync::atomic::AtomicBool::new(false));
        let ran_hook = StdArc::clone(&ran);
        let err = prepare_url_target_with_resolver(
            "http://127.0.0.1/",
            deadline,
            move |_| {
                ran_hook.store(true, Ordering::SeqCst);
                std::thread::sleep(Duration::from_millis(80));
                Ok(())
            },
            &fake,
        )
        .await
        .expect_err("checkpoint 4 catches post-hook expiry");
        assert!(matches!(err, UrlTargetError::DeadlineExceeded));
        assert!(
            ran.load(Ordering::SeqCst),
            "the synchronous hook still ran to completion"
        );
    }

    #[tokio::test]
    async fn public_wrapper_prepares_a_literal_target() {
        let target = prepare_url_target("http://127.0.0.1:9/health", far_deadline(), |_| Ok(()))
            .await
            .expect("literal prepares");
        assert!(!target.is_tls());
        assert_eq!(target.addresses(), ["127.0.0.1:9".parse::<SocketAddr>().unwrap()]);
        assert_eq!(target.host_authority().to_str().unwrap(), "127.0.0.1:9");
    }

    async fn spawn_loopback_backend(body: &'static str) -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        tokio::spawn(async move {
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    return;
                };
                let response = body.to_owned();
                tokio::spawn(async move {
                    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
                    let mut buf = [0_u8; 1024];
                    drop(stream.read(&mut buf).await);
                    let msg = format!(
                        "HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
                        response.len(),
                        response
                    );
                    drop(stream.write_all(msg.as_bytes()).await);
                    drop(stream.flush().await);
                });
            }
        });
        port
    }

    fn get_request() -> SubRequest {
        SubRequest {
            method: http::Method::GET,
            uri: "/".parse().unwrap(),
            headers: http::HeaderMap::new(),
            body: bytes::Bytes::new(),
        }
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "buffered + streaming parity in one test proves both share the prepared target"
    )]
    async fn buffered_and_streaming_share_the_prepared_target() {
        let port = spawn_loopback_backend("hello-parity").await;
        let url = format!("http://127.0.0.1:{port}/");
        let deadline = Instant::now() + Duration::from_secs(5);
        // Direct construction misses the bootstrap that installs the provider.
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));

        let target = prepare_url_target(&url, deadline, |_| Ok(())).await.unwrap();
        let prepared = target.bind(get_request());
        let peer = prepared.peer_at(0).expect("one address");
        let resp = Box::pin(client.execute(&peer, prepared.request(), 1_048_576, Duration::from_secs(5), None))
            .await
            .expect("buffered exchange");
        assert_eq!(resp.status, 200);
        assert_eq!(resp.body, bytes::Bytes::from_static(b"hello-parity"));

        let target = prepare_url_target(&url, deadline, |_| Ok(())).await.unwrap();
        let prepared = target.bind(get_request());
        let peer = prepared.peer_at(0).expect("one address");
        let limits = StreamLimits {
            idle_timeout: Duration::from_secs(5),
            max_stream_duration: None,
            max_total_bytes: None,
        };
        let mut streaming =
            Box::pin(client.send_streaming(&peer, prepared.request(), Duration::from_secs(5), limits, None))
                .await
                .expect("streaming exchange");
        assert_eq!(streaming.status, 200);
        let mut collected = Vec::new();
        while let Some(chunk) = streaming.body.next_chunk().await.expect("chunk") {
            collected.extend_from_slice(&chunk);
        }
        assert_eq!(collected, b"hello-parity");
        drop(streaming);
    }

    #[tokio::test]
    #[expect(clippy::too_many_lines, reason = "checks address fallback and authority on the wire")]
    async fn execute_url_uses_frozen_fallback_and_url_authority() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = [0_u8; 2048];
            let count = stream.read(&mut buf).await.unwrap();
            stream
                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok")
                .await
                .unwrap();
            String::from_utf8_lossy(&buf[..count]).into_owned()
        });
        let host = "url-fallback.praxis-test.invalid";
        crate::connectivity::peer::seed_dns(host, &["::1".parse().unwrap(), "127.0.0.1".parse().unwrap()]);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let mut request = req_with_host("wrong.example");
        request.uri = "/wrong".parse().unwrap();
        let url = format!("http://{host}:{port}/right?token=secret-url-token");
        let mut seen = Vec::new();
        let response = Box::pin(client.execute_url(
            &url,
            request,
            |addrs| {
                seen.extend_from_slice(addrs);
                // Poison the cache with a single bad answer AFTER the full set was
                // resolved and validated. A correct implementation dials the
                // already-frozen peers and never re-resolves, so the healthy
                // 127.0.0.1 fallback still succeeds; a re-resolving one would see
                // only ::1 here and fail.
                crate::connectivity::peer::seed_dns(host, &["::1".parse().unwrap()]);
                Ok(())
            },
            UrlResolutionPolicy::OperatorCached,
            Duration::from_secs(5),
            1024,
            None,
        ))
        .await
        .unwrap();
        assert_eq!(
            response.body.as_ref(),
            b"ok",
            "frozen fallback must reach 127.0.0.1 even after the cache is poisoned mid-call"
        );
        assert_eq!(seen.len(), 2, "validation sees every address before either dial");
        assert_eq!(
            seen[0].ip(),
            "::1".parse::<IpAddr>().unwrap(),
            "validator must see the first DNS answer"
        );
        assert_eq!(
            seen[1].ip(),
            "127.0.0.1".parse::<IpAddr>().unwrap(),
            "validator must see the fallback DNS answer"
        );
        let captured = backend.await.unwrap();
        let captured_lower = captured.to_ascii_lowercase();
        assert!(
            captured.starts_with("GET /right?token=secret-url-token HTTP/1.1"),
            "{captured}"
        );
        assert!(
            captured_lower.contains(&format!("host: {host}:{port}\r\n")),
            "{captured}"
        );
        assert!(
            !captured.contains("wrong.example"),
            "URL authority must replace the caller's Host header: {captured}"
        );
    }

    #[tokio::test]
    #[expect(clippy::too_many_lines, reason = "exercises the all-peers-fail redaction end to end")]
    async fn execute_url_all_addresses_fail_returns_redacted_connect_error() {
        // Bind then drop to obtain a port nothing listens on, so every dial is
        // refused immediately rather than hanging.
        let closed = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = closed.local_addr().unwrap().port();
        drop(closed);
        let host = "url-all-fail.praxis-test.invalid";
        // Both loopback interfaces are configured by default and refuse a closed
        // port immediately; 127.0.0.2 would hang on macOS instead of refusing.
        crate::connectivity::peer::seed_dns(host, &["127.0.0.1".parse().unwrap(), "::1".parse().unwrap()]);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://{host}:{port}/?token=secret-url-token");
        let err = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::OperatorCached,
            Duration::from_secs(5),
            1024,
            None,
        ))
        .await
        .unwrap_err();
        assert!(
            matches!(err, UrlSubRequestError::Exchange(SubRequestError::Connect(_))),
            "every address refusing must surface as a typed Connect error: {err:?}"
        );
        assert_eq!(
            err.to_string(),
            "sub-request connect error: upstream connection failed",
            "the all-addresses-fail error must carry the fixed, redacted connect text"
        );
        assert!(
            !format!("{err:?}").contains("secret-url-token"),
            "Debug of the all-addresses-fail error must not leak the URL query: {err:?}"
        );
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "tests two requests against an open and a healthy circuit"
    )]
    async fn execute_url_skips_open_circuit_and_reaches_healthy_peer() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            for _ in 0..2 {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut buf = [0_u8; 1024];
                drop(stream.read(&mut buf).await);
                stream
                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\nconnection: close\r\n\r\nok")
                    .await
                    .unwrap();
            }
        });
        let host = "url-circuit-fallback.praxis-test.invalid";
        crate::connectivity::peer::seed_dns(host, &["::1".parse().unwrap(), "127.0.0.1".parse().unwrap()]);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::with_options(SubRequestConnectorOptions {
            keepalive_pool_size: 1,
            max_connections: Some(2),
            circuit_breaker: Some(crate::circuit::CircuitBreakerConfig {
                threshold: 1,
                recovery_window: Duration::from_secs(30),
                half_open_timeout: Duration::from_secs(30),
            }),
        }));
        let url = format!("http://{host}:{port}/");
        for _ in 0..2 {
            let response = Box::pin(client.execute_url(
                &url,
                get_request(),
                |_| Ok(()),
                UrlResolutionPolicy::OperatorCached,
                Duration::from_secs(5),
                1024,
                None,
            ))
            .await
            .expect("healthy address remains reachable after the first peer's circuit opens");
            assert_eq!(
                response.body.as_ref(),
                b"ok",
                "healthy peer must answer despite the other open circuit"
            );
        }
        backend.await.unwrap();
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "tests typed error and redaction through an upstream response"
    )]
    async fn execute_url_redacts_reflected_query_in_upstream_error() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = [0_u8; 1024];
            let count = stream.read(&mut buf).await.unwrap();
            let request = String::from_utf8_lossy(&buf[..count]);
            let target = request.split_whitespace().nth(1).unwrap();
            let malformed = format!("HTTP/1.1 200 OK\r\nLocation: {target}\r\nbad header: value\r\n\r\n");
            stream.write_all(malformed.as_bytes()).await.unwrap();
        });
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://127.0.0.1:{port}/?token=secret-url-token");
        let err = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_secs(5),
            1024,
            None,
        ))
        .await
        .unwrap_err();
        assert!(
            matches!(err, UrlSubRequestError::Exchange(SubRequestError::Io(_))),
            "malformed upstream response must surface as I/O failure: {err:?}"
        );
        assert!(
            !err.to_string().contains("secret-url-token"),
            "Display must redact the URL query: {err}"
        );
        assert!(
            !format!("{err:?}").contains("secret-url-token"),
            "Debug must redact the URL query: {err:?}"
        );
        backend.await.unwrap();
    }

    #[tokio::test]
    #[expect(clippy::too_many_lines, reason = "verifies full-set rejection before any dial")]
    async fn execute_url_rejection_prevents_every_dial_and_hides_query() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let host = "url-reject.praxis-test.invalid";
        crate::connectivity::peer::seed_dns(host, &["127.0.0.1".parse().unwrap(), "10.0.0.1".parse().unwrap()]);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://{host}:{port}/?token=secret-url-token");
        let err = Box::pin(client.execute_url(
            &url,
            get_request(),
            |addrs| {
                assert_eq!(addrs.len(), 2, "policy must inspect the full resolved set");
                Err("private address rejected".into())
            },
            UrlResolutionPolicy::OperatorCached,
            Duration::from_secs(5),
            1024,
            None,
        ))
        .await
        .unwrap_err();
        assert!(
            matches!(err, UrlSubRequestError::Target(UrlTargetError::PolicyRejected(_))),
            "policy rejection must remain a typed target error: {err:?}"
        );
        assert!(
            !err.to_string().contains("secret-url-token"),
            "policy error must redact the URL query: {err}"
        );
        assert!(
            tokio::time::timeout(Duration::from_millis(50), listener.accept())
                .await
                .is_err(),
            "validation rejection must prevent all connection attempts"
        );
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "tests that response I/O does not trigger address fallback"
    )]
    async fn execute_url_does_not_fall_back_after_response_io_failure() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = [0_u8; 1024];
            drop(stream.read(&mut buf).await);
            stream
                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 4\r\nconnection: close\r\n\r\nxx")
                .await
                .unwrap();
        });
        let host = "url-no-io-fallback.praxis-test.invalid";
        crate::connectivity::peer::seed_dns(host, &["127.0.0.1".parse().unwrap(), "127.0.0.2".parse().unwrap()]);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://{host}:{port}/");
        let result = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::OperatorCached,
            Duration::from_secs(5),
            1024,
            None,
        ))
        .await;
        assert!(
            matches!(result, Err(UrlSubRequestError::Exchange(SubRequestError::Io(_)))),
            "an I/O failure after connecting must not fall back: {result:?}"
        );
        backend.await.unwrap();
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "tests response size and deadline on successive exchanges"
    )]
    async fn execute_url_enforces_response_limit_and_single_deadline() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            for stall in [false, true] {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut buf = [0_u8; 1024];
                drop(stream.read(&mut buf).await);
                stream
                    .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 8\r\n\r\n")
                    .await
                    .unwrap();
                if stall {
                    tokio::time::sleep(Duration::from_secs(1)).await;
                } else {
                    stream.write_all(b"12345678").await.unwrap();
                }
            }
        });
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://127.0.0.1:{port}/");
        let too_large = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_secs(5),
            4,
            None,
        ))
        .await;
        assert!(
            matches!(
                too_large,
                Err(UrlSubRequestError::Exchange(SubRequestError::ResponseTooLarge {
                    limit: 4,
                    ..
                }))
            ),
            "response body must obey the per-call four-byte limit: {too_large:?}"
        );
        let timed_out = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_millis(50),
            1024,
            None,
        ))
        .await;
        assert!(
            matches!(timed_out, Err(UrlSubRequestError::DeadlineExceeded)),
            "response wait must obey the single URL deadline: {timed_out:?}"
        );
        backend.abort();
    }

    #[tokio::test]
    #[expect(
        clippy::too_many_lines,
        reason = "end-to-end timing proof: validation spends the budget that I/O then cannot"
    )]
    #[allow(
        clippy::disallowed_methods,
        reason = "the synchronous validate hook intentionally burns real wall-clock budget"
    )]
    async fn execute_url_charges_validation_time_to_single_deadline() {
        use tokio::io::AsyncReadExt as _;

        // A backend that connects but never answers, so the only way the call can
        // return is the overall deadline elapsing during the response wait.
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let backend = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = [0_u8; 1024];
            drop(stream.read(&mut buf).await);
            tokio::time::sleep(Duration::from_secs(30)).await;
        });
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let url = format!("http://127.0.0.1:{port}/");
        let started = Instant::now();
        let result = Box::pin(client.execute_url(
            &url,
            get_request(),
            |_| {
                // Burn ~400ms of a 500ms budget before any dial. If the deadline
                // restarted after preparation, the stalled response read would run
                // the full 500ms and total ~900ms; the single clock keeps it short.
                std::thread::sleep(Duration::from_millis(400));
                Ok(())
            },
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_millis(500),
            1024,
            None,
        ))
        .await;
        let elapsed = started.elapsed();
        assert!(
            matches!(result, Err(UrlSubRequestError::DeadlineExceeded)),
            "time spent validating must shrink the later exchange budget: {result:?}"
        );
        assert!(
            elapsed < Duration::from_millis(850),
            "the single deadline must span validation and I/O, not restart per stage: {elapsed:?}"
        );
        backend.abort();
    }

    #[tokio::test]
    async fn execute_url_rejects_framework_host_override() {
        let mut headers = crate::subrequest::FrameworkHeaders::new();
        headers
            .insert(http::header::HOST, http::HeaderValue::from_static("wrong.example"))
            .unwrap();
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let result = Box::pin(client.execute_url(
            "http://127.0.0.1:9/",
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_secs(1),
            1024,
            Some(&headers),
        ))
        .await;
        assert!(
            matches!(
                result,
                Err(UrlSubRequestError::Exchange(SubRequestError::InvalidRequest(_)))
            ),
            "framework headers must not override the URL Host authority: {result:?}"
        );
    }

    #[tokio::test]
    async fn execute_url_rejects_framework_host_removal() {
        // A framework Host *removal* must be rejected exactly like an override:
        // without this guard the bound URL authority would be stripped and the
        // peer would receive a bare `IP:port` Host instead.
        let mut headers = crate::subrequest::FrameworkHeaders::new();
        headers.remove(http::header::HOST);
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let result = Box::pin(client.execute_url(
            "http://127.0.0.1:9/",
            get_request(),
            |_| Ok(()),
            UrlResolutionPolicy::ClientPerCall,
            Duration::from_secs(1),
            1024,
            Some(&headers),
        ))
        .await;
        assert!(
            matches!(
                result,
                Err(UrlSubRequestError::Exchange(SubRequestError::InvalidRequest(_)))
            ),
            "framework headers must not remove the URL Host authority: {result:?}"
        );
    }

    #[tokio::test]
    #[expect(clippy::too_many_lines, reason = "tests explicit-peer Host on the wire")]
    async fn explicit_peer_execute_preserves_virtual_host() {
        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let backend = tokio::spawn(async move {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut buf = [0_u8; 1024];
            let count = stream.read(&mut buf).await.unwrap();
            stream
                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n")
                .await
                .unwrap();
            String::from_utf8_lossy(&buf[..count]).into_owned()
        });
        praxis_tls::provider::install();
        let client = SubRequestClient::new(SubRequestConnector::new(1, None));
        let peer = HttpPeer::new(addr, false, String::new());
        let response = Box::pin(client.execute(
            &peer,
            &req_with_host("virtual.example"),
            1024,
            Duration::from_secs(5),
            None,
        ))
        .await
        .unwrap();
        assert_eq!(
            response.status, 200,
            "explicit-peer execution must retain a successful response"
        );
        let captured = backend.await.unwrap();
        assert!(
            captured.to_ascii_lowercase().contains("host: virtual.example\r\n"),
            "{captured}"
        );
    }

    #[test]
    fn url_target_error_debug() {
        let err = UrlTargetError::DeadlineExceeded;
        let debug = format!("{err:?}");
        assert!(debug.contains("DeadlineExceeded"));

        let invalid = UrlTargetError::InvalidTarget(InvalidTarget::MissingHost);
        let debug = format!("{invalid:?}");
        assert!(debug.contains("InvalidTarget"));
        assert!(debug.contains("MissingHost"));

        let resolve = UrlTargetError::Resolve(AddressResolutionError::Empty("test".to_owned()));
        let debug = format!("{resolve:?}");
        assert!(debug.contains("Resolve"));
        assert!(debug.contains("Empty"));
    }

    #[test]
    fn url_target_error_display() {
        assert_eq!(
            UrlTargetError::DeadlineExceeded.to_string(),
            "target preparation deadline exceeded"
        );

        let err = UrlTargetError::InvalidTarget(InvalidTarget::MissingHost);
        assert_eq!(err.to_string(), "target has no host");

        let policy = UrlTargetError::PolicyRejected("test".into());
        assert_eq!(policy.to_string(), "address policy rejected the resolved target");
    }

    #[test]
    fn invalid_target_debug() {
        let err = InvalidTarget::MissingHost;
        let debug = format!("{err:?}");
        assert!(debug.contains("MissingHost"));

        let malformed = InvalidTarget::Malformed("test error".to_owned());
        let debug = format!("{malformed:?}");
        assert!(debug.contains("Malformed"));
        assert!(debug.contains("test error"));

        let scheme = InvalidTarget::UnsupportedScheme("ws".to_owned());
        let debug = format!("{scheme:?}");
        assert!(debug.contains("UnsupportedScheme"));
        assert!(debug.contains("ws"));
    }

    #[test]
    fn parsed_target_debug_redacts_query() {
        let p = parse_ok("https://api.example.com/v1/models?token=secret123&key=value");
        let debug = format!("{p:?}");
        assert!(debug.contains("ParsedTarget"));
        assert!(debug.contains("is_tls"));
        assert!(debug.contains("true"));
        assert!(debug.contains("api.example.com"));
        assert!(debug.contains("/v1/models"), "path should be visible");
        assert!(
            !debug.contains("token=secret123"),
            "query should be redacted from Debug"
        );
        assert!(!debug.contains("key=value"), "query should be redacted from Debug");
    }

    #[test]
    fn prepared_target_debug_redacts_query() {
        let t = target(true, "api.example.com:443", "api.example.com", &["93.184.216.34:443"]);
        let debug = format!("{t:?}");
        assert!(debug.contains("PreparedTarget"));
        assert!(debug.contains("is_tls"));
        assert!(debug.contains("true"));
        assert!(debug.contains("api.example.com"));
        assert!(debug.contains("/p"), "path should be visible");
        assert!(!debug.contains("q=1"), "query should be redacted from Debug");
    }

    #[test]
    fn validate_url_target_success() {
        assert!(validate_url_target("http://example.com/").is_ok());
        assert!(validate_url_target("https://api.example.com:8443/path").is_ok());
        assert!(validate_url_target("http://127.0.0.1:9000/").is_ok());
        assert!(validate_url_target("https://[::1]/").is_ok());
    }

    #[test]
    fn validate_url_target_failures() {
        assert!(validate_url_target("ftp://example.com/").is_err());
        assert!(validate_url_target("http://user@host/").is_err());
        assert!(validate_url_target("http://host/#fragment").is_err());
        assert!(validate_url_target("http://:80/").is_err());
        assert!(validate_url_target("http://host:99999/").is_err());
    }

    #[tokio::test]
    async fn deadline_checkpoint_after_parse() {
        tokio::time::pause();
        let deadline = Instant::now() + Duration::from_millis(10);
        tokio::time::advance(Duration::from_millis(11)).await;
        let fake = FakeResolver::ok(vec![]);
        let err = prepare_url_target_with_resolver("http://127.0.0.1/", deadline, |_| Ok(()), &fake)
            .await
            .expect_err("checkpoint 2 should fire");
        assert!(matches!(err, UrlTargetError::DeadlineExceeded));
        assert_eq!(fake.call_count(), 0, "checkpoint 2 happens before DNS");
    }

    #[tokio::test]
    async fn prepared_subrequest_addresses_accessor() {
        let t = target(false, "h", "", &["10.0.0.1:80", "10.0.0.2:80"]);
        let prepared = t.bind(req_with_host("ignored"));
        let addrs = prepared.addresses();
        assert_eq!(addrs.len(), 2);
        assert_eq!(addrs[0].to_string(), "10.0.0.1:80");
        assert_eq!(addrs[1].to_string(), "10.0.0.2:80");
    }

    #[tokio::test]
    async fn prepared_subrequest_request_accessor() {
        let t = target(true, "api.example.com:443", "api.example.com", &["93.184.216.34:443"]);
        let req = req_with_host("original.example");
        let prepared = t.bind(req);
        let bound_req = prepared.request();
        assert_eq!(
            bound_req.headers.get(http::header::HOST).unwrap().to_str().unwrap(),
            "api.example.com:443",
            "bound request should have URL authority as Host"
        );
    }

    #[tokio::test]
    async fn system_resolver_error_path() {
        let result = SystemResolver
            .resolve_host("nonexistent-host-12345-abcde.invalid")
            .await;
        assert!(result.is_err(), "nonexistent domain should fail to resolve");
    }

    #[test]
    fn parses_port_65535() {
        let p = parse_ok("http://example.com:65535/");
        assert_eq!(p.effective_port, 65535);
    }

    #[test]
    fn parses_ipv6_with_zone_id_rejected() {
        let err = parse_target("http://[fe80::1%eth0]/").unwrap_err();
        assert!(
            matches!(err, InvalidTarget::InvalidHost(_)),
            "zone IDs should be rejected: {err:?}"
        );
    }

    #[test]
    fn rejects_missing_scheme() {
        let err = parse_target("example.com/").unwrap_err();
        assert!(
            matches!(err, InvalidTarget::Malformed(_)),
            "missing scheme should be Malformed"
        );
    }

    #[test]
    fn rejects_unterminated_bracket() {
        let err = parse_target("http://[::1/").unwrap_err();
        assert!(
            matches!(err, InvalidTarget::Malformed(_)),
            "unterminated bracket should be Malformed: {err:?}"
        );
    }

    #[test]
    fn address_resolution_error_converts() {
        let resolve_err = AddressResolutionError::Empty("test".to_owned());
        let url_err: UrlTargetError = resolve_err.into();
        assert!(matches!(url_err, UrlTargetError::Resolve(_)));
    }

    #[test]
    fn prepared_target_new_with_http_clears_sni() {
        let t = PreparedTarget::new(
            false,
            http::HeaderValue::from_str("example.com").unwrap(),
            "example.com".to_owned(),
            "/".parse().unwrap(),
            vec!["93.184.216.34:80".parse().unwrap()],
        );
        assert_eq!(t.sni(), "", "http target should have empty SNI even if passed");
    }

    #[test]
    fn prepared_target_new_with_https_keeps_sni() {
        let t = PreparedTarget::new(
            true,
            http::HeaderValue::from_str("example.com:443").unwrap(),
            "example.com".to_owned(),
            "/".parse().unwrap(),
            vec!["93.184.216.34:443".parse().unwrap()],
        );
        assert_eq!(t.sni(), "example.com");
    }

    #[tokio::test]
    async fn multiple_addresses_all_reachable_via_peer_at() {
        let t = target(
            true,
            "h:443",
            "h.example.com",
            &["10.0.0.1:443", "10.0.0.2:443", "10.0.0.3:443"],
        );
        let prepared = t.bind(req_with_host("ignored"));
        assert_eq!(prepared.peer_at(0).unwrap().address().to_string(), "10.0.0.1:443");
        assert_eq!(prepared.peer_at(1).unwrap().address().to_string(), "10.0.0.2:443");
        assert_eq!(prepared.peer_at(2).unwrap().address().to_string(), "10.0.0.3:443");
        assert!(prepared.peer_at(3).is_none());
        assert!(prepared.peer_at(999).is_none());
    }

    #[test]
    fn parses_query_with_multiple_params() {
        let p = parse_ok("https://api.example.com/endpoint?a=1&b=2&c=3");
        assert_eq!(p.origin_form.to_string(), "/endpoint?a=1&b=2&c=3");
    }

    #[test]
    fn parses_encoded_characters_in_path() {
        let p = parse_ok("https://example.com/path%20with%20spaces");
        assert_eq!(p.origin_form.to_string(), "/path%20with%20spaces");
    }

    #[tokio::test]
    async fn policy_rejected_preserves_source() {
        let source: Box<dyn std::error::Error + Send + Sync> = "policy violation".into();
        let err = UrlTargetError::PolicyRejected(source);
        assert!(
            std::error::Error::source(&err).is_some(),
            "PolicyRejected should preserve source for programmatic access"
        );
    }
}