zlayer_agent/

proxy_manager.rs

//! Proxy management for agent-controlled services
//!
//! This module provides the `ProxyManager` struct that integrates the proxy crate
//! with the agent's service management. It handles:
//! - Managing proxy routes based on `ServiceSpec` endpoints (HTTP/HTTPS/WebSocket)
//! - Managing L4 stream proxy listeners (TCP/UDP)
//! - Tracking and updating backend servers for load balancing
//! - Coordinating proxy server lifecycle

use crate::error::Result;
use std::collections::{HashMap, HashSet};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use zlayer_proxy::{
    endpoint_lb_key, load_existing_certs_into_resolver, tls_acceptor_from_resolver, Activator,
    CertManager, LbStrategy, LoadBalancer, NetworkPolicyChecker, ProxyConfig, ProxyServer,
    RouteEntry, RpsRegistry, ServiceRegistry, SniCertResolver, StreamHealthProbe,
    StreamProxyConfig, StreamRegistry, StreamService, TcpStreamService, UdpStreamService,
};
use zlayer_scheduler::scalers::RpsProvider;
use zlayer_spec::{
    ExposeType, PortMapping, PortProtocol, Protocol, ServiceSpec, StreamEndpointConfig,
    StreamHealthCheck,
};

/// Default activation floor: the replica count a [`ServiceActivator`] scales a
/// scaled-to-zero service up to on the first request.
const DEFAULT_ACTIVATION_FLOOR: u32 = 1;

/// Bind an explicitly IPv6-only TCP listener on `[::]:port`.
///
/// The socket is forced `IPV6_V6ONLY` so it never conflicts with the sibling
/// `0.0.0.0:port` IPv4 listener — on Linux the default `IPV6_V6ONLY=0` would
/// otherwise make `[::]` claim v4-mapped traffic too and the second bind would
/// fail with `EADDRINUSE`. Returned as a non-blocking [`tokio::net::TcpListener`]
/// ready to hand to the proxy accept loop.
fn bind_v6_only(port: u16) -> std::io::Result<tokio::net::TcpListener> {
    use socket2::{Domain, Protocol, Socket, Type};
    let sock = Socket::new(Domain::IPV6, Type::STREAM, Some(Protocol::TCP))?;
    sock.set_only_v6(true)?;
    sock.set_reuse_address(true)?;
    sock.set_nonblocking(true)?;
    let addr: std::net::SocketAddr = (std::net::Ipv6Addr::UNSPECIFIED, port).into();
    sock.bind(&addr.into())?;
    sock.listen(1024)?;
    tokio::net::TcpListener::from_std(sock.into())
}

/// Translate an endpoint's `stream:` block (the controlling
/// [`StreamEndpointConfig`] from `zlayer-types`) into the proxy crate's
/// runtime [`StreamProxyConfig`].
///
/// This is where the spec-level types are decoded into proxy-local ones:
/// `session_timeout` strings are parsed into [`Duration`], and a configured
/// [`StreamHealthCheck`] is lowered into a [`StreamHealthProbe`] (decoding the
/// `\xNN` hex escapes in the UDP probe request/expect payloads).
///
/// `None` (no `stream:` block) yields the default config (no TLS, no PROXY
/// protocol, no session-timeout override, no health probe).
fn translate_stream_config(stream: Option<&StreamEndpointConfig>) -> StreamProxyConfig {
    let Some(stream) = stream else {
        return StreamProxyConfig::default();
    };

    let health_check = stream.health_check.as_ref().map(|hc| match hc {
        StreamHealthCheck::TcpConnect => StreamHealthProbe::TcpConnect,
        StreamHealthCheck::UdpProbe { request, expect } => StreamHealthProbe::UdpProbe {
            request: unescape_hex(request),
            expect: expect.as_deref().map(unescape_hex),
        },
    });

    StreamProxyConfig {
        tls: stream.tls,
        proxy_protocol: stream.proxy_protocol,
        session_timeout: stream.session_timeout_duration(),
        health_check,
    }
}

/// Decode `\xNN` hex escapes (and `\\` for a literal backslash) in a probe
/// payload string into raw bytes. Any other character (or a malformed escape)
/// is passed through as its UTF-8 bytes, so plain ASCII payloads work as-is.
fn unescape_hex(s: &str) -> Vec<u8> {
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'\\' && i + 1 < bytes.len() {
            match bytes[i + 1] {
                b'x' | b'X' if i + 3 < bytes.len() => {
                    let hi = (bytes[i + 2] as char).to_digit(16);
                    let lo = (bytes[i + 3] as char).to_digit(16);
                    if let (Some(hi), Some(lo)) = (hi, lo) {
                        // hi/lo are each a single hex digit (0..=15), so the
                        // combined value is 0..=255 — try_from never fails.
                        out.push(u8::try_from((hi << 4) | lo).unwrap_or(0));
                        i += 4;
                        continue;
                    }
                    // Malformed \x escape: emit the backslash literally.
                    out.push(b'\\');
                    i += 1;
                }
                b'\\' => {
                    out.push(b'\\');
                    i += 2;
                }
                _ => {
                    out.push(b'\\');
                    i += 1;
                }
            }
        } else {
            out.push(bytes[i]);
            i += 1;
        }
    }
    out
}

/// Maximum time [`ServiceActivator::activate`] waits for a healthy backend to
/// appear after triggering a scale-up before returning so the proxy can fall
/// back to its own deadline/`503`.
const ACTIVATOR_READY_DEADLINE: Duration = Duration::from_secs(30);

/// Poll step while waiting for a backend after a scale-up trigger.
const ACTIVATOR_POLL_STEP: Duration = Duration::from_millis(200);

/// Abstraction over "scale this service to N replicas", implemented by the
/// agent's `ServiceManager`.
///
/// [`ProxyManager`] is constructed *before* the `ServiceManager` and the
/// `ServiceManager` holds an `Arc<ProxyManager>` (one-way wiring), so the
/// activator cannot hold a concrete `ServiceManager` without a reference cycle.
/// This trait lets the activator depend only on the scale-up capability, which
/// the `ServiceManager` provides via its existing `scale_service` method. The
/// daemon installs a concrete trigger after both halves exist
/// (see [`ProxyManager::set_activator`]).
#[async_trait::async_trait]
pub trait ScaleTrigger: Send + Sync {
    /// Scale `service` to at least `replicas` running replicas.
    ///
    /// # Errors
    ///
    /// Returns a human-readable error if the scale-up could not be initiated.
    async fn scale_to(&self, service: &str, replicas: u32) -> std::result::Result<(), String>;
}

/// On-demand [`Activator`] for scale-to-zero services.
///
/// When the proxy resolves a route with no healthy backend, it calls
/// [`Activator::activate`] with the resolved LB group key. This implementation
/// derives the bare service name from that key, triggers a scale-up to the
/// activation floor via a [`ScaleTrigger`], then polls the shared
/// [`LoadBalancer`] until a healthy backend for the key appears (or the
/// deadline elapses).
pub struct ServiceActivator {
    /// The scale-up capability (the agent's `ServiceManager`).
    trigger: Arc<dyn ScaleTrigger>,
    /// Shared load balancer the proxy selects from; polled for readiness.
    load_balancer: Arc<LoadBalancer>,
    /// Replica floor to scale up to on activation.
    floor: u32,
}

impl ServiceActivator {
    /// Create an activator that scales up via `trigger` and waits on
    /// `load_balancer` for readiness, using [`DEFAULT_ACTIVATION_FLOOR`].
    #[must_use]
    pub fn new(trigger: Arc<dyn ScaleTrigger>, load_balancer: Arc<LoadBalancer>) -> Self {
        Self {
            trigger,
            load_balancer,
            floor: DEFAULT_ACTIVATION_FLOOR,
        }
    }

    /// Override the activation replica floor (default
    /// [`DEFAULT_ACTIVATION_FLOOR`]). A floor of `0` is clamped to `1` so an
    /// activation always brings the service up.
    #[must_use]
    pub fn with_floor(mut self, floor: u32) -> Self {
        self.floor = floor.max(1);
        self
    }

    /// Derive the bare service name from a resolved LB group key.
    ///
    /// `add_service` registers per-endpoint LB groups under the composite key
    /// `{deployment}/{service}#{endpoint}` (and a bare service-level group).
    /// `scale_service` operates on the bare service name, so strip the
    /// `#endpoint` suffix and the `deployment/` prefix.
    fn service_name_from_key(key: &str) -> &str {
        let without_endpoint = key.split('#').next().unwrap_or(key);
        without_endpoint
            .rsplit('/')
            .next()
            .unwrap_or(without_endpoint)
    }
}

#[async_trait::async_trait]
impl Activator for ServiceActivator {
    async fn activate(&self, service: &str) -> std::result::Result<(), String> {
        let bare = Self::service_name_from_key(service);
        info!(
            lb_key = %service,
            service = %bare,
            floor = self.floor,
            "Activating scaled-to-zero service on demand"
        );

        // Trigger the scale-up. A failure here is surfaced to the proxy, which
        // still polls for a backend (a concurrent activation may bring it up).
        self.trigger.scale_to(bare, self.floor).await?;

        // Wait for a healthy backend on the SAME key the proxy selects with.
        let deadline = Instant::now() + ACTIVATOR_READY_DEADLINE;
        loop {
            if self.load_balancer.select(service).is_some() {
                return Ok(());
            }
            if Instant::now() >= deadline {
                return Err(format!(
                    "service '{bare}' did not become ready within {ACTIVATOR_READY_DEADLINE:?} after scale-up"
                ));
            }
            tokio::time::sleep(ACTIVATOR_POLL_STEP).await;
        }
    }
}

/// Adapter exposing a [`RpsRegistry`] as a scheduler [`RpsProvider`].
///
/// The `zlayer-proxy` crate must not depend on `zlayer-scheduler`, so the
/// `RpsProvider` impl lives here in the agent (which depends on both). The
/// scheduler reads per-service RPS through this adapter; the proxy populates the
/// underlying registry on every routed request.
#[derive(Debug, Clone)]
pub struct RpsRegistryProvider {
    /// The registry populated by the proxy request path.
    registry: Arc<RpsRegistry>,
}

impl RpsRegistryProvider {
    /// Wrap a shared [`RpsRegistry`] as an [`RpsProvider`].
    #[must_use]
    pub fn new(registry: Arc<RpsRegistry>) -> Self {
        Self { registry }
    }
}

#[async_trait::async_trait]
impl RpsProvider for RpsRegistryProvider {
    async fn rps(&self, service: &str) -> f64 {
        self.registry.rps(service).await
    }
}

/// Default HTTP ingress port. The lone daemon IS the cluster ingress: it binds
/// `0.0.0.0:80` so any deployed service with an HTTP endpoint is reachable on
/// the standard web port without per-service listener configuration.
pub const DEFAULT_INGRESS_HTTP_PORT: u16 = 80;

/// Default HTTPS ingress port. See [`DEFAULT_INGRESS_HTTP_PORT`].
pub const DEFAULT_INGRESS_HTTPS_PORT: u16 = 443;

/// Configuration for the `ProxyManager`
#[derive(Debug, Clone)]
pub struct ProxyManagerConfig {
    /// HTTP bind address
    pub http_addr: SocketAddr,
    /// HTTPS bind address (optional)
    pub https_addr: Option<SocketAddr>,
    /// Whether to enable HTTP/2
    pub http2_enabled: bool,
}

impl Default for ProxyManagerConfig {
    fn default() -> Self {
        Self {
            http_addr: "0.0.0.0:80".parse().unwrap(),
            https_addr: None,
            http2_enabled: true,
        }
    }
}

impl ProxyManagerConfig {
    /// Create a new configuration with the specified HTTP address
    #[must_use]
    pub fn new(http_addr: SocketAddr) -> Self {
        Self {
            http_addr,
            https_addr: None,
            http2_enabled: true,
        }
    }

    /// Set the HTTPS address
    #[must_use]
    pub fn with_https(mut self, addr: SocketAddr) -> Self {
        self.https_addr = Some(addr);
        self
    }

    /// Set HTTP/2 support
    #[must_use]
    pub fn with_http2(mut self, enabled: bool) -> Self {
        self.http2_enabled = enabled;
        self
    }
}

/// Per-service tracking information for cleanup purposes.
#[derive(Debug, Clone)]
struct ServiceTracking {
    /// Owning deployment name, when known. Threaded from
    /// `ServiceSpec.deployment` by `add_service`. `None` for standalone /
    /// single-deployment callers (`docker run`). Used to build the
    /// deployment-scoped LB group key so two deployments sharing a
    /// service+endpoint name keep independent backend pools.
    deployment: Option<String>,
    /// Endpoint names (used to derive per-endpoint LB group keys for
    /// cleanup on `remove_service`).
    endpoint_names: Vec<String>,
    /// TCP ports owned by this service
    tcp_ports: Vec<u16>,
    /// UDP ports owned by this service
    udp_ports: Vec<u16>,
    /// HTTP/HTTPS/WebSocket ports owned by this service
    http_ports: Vec<u16>,
}

/// Manages proxy routing for agent-controlled services
///
/// The `ProxyManager` coordinates between the agent's service lifecycle and
/// the proxy crate's routing/load balancing infrastructure. It supports:
///
/// - **HTTP/HTTPS/WebSocket (L7)**: Multiple port listeners sharing the same
///   `ServiceRegistry` for request matching and load balancing.
/// - **TCP/UDP (L4)**: Standalone stream proxy listeners that forward raw
///   connections/datagrams to backends via the `StreamRegistry`.
pub struct ProxyManager {
    /// Configuration
    config: ProxyManagerConfig,
    /// Shared service registry for HTTP request matching and backend management
    registry: Arc<ServiceRegistry>,
    /// Load balancer for health-aware backend selection
    load_balancer: Arc<LoadBalancer>,
    /// Per-port HTTP proxy server handles
    servers: RwLock<HashMap<u16, Arc<ProxyServer>>>,
    /// Tracked services and their endpoints (includes port ownership for cleanup)
    services: RwLock<HashMap<String, ServiceTracking>>,
    /// Stream registry for L4 TCP/UDP proxy routing
    stream_registry: Option<Arc<StreamRegistry>>,
    /// Certificate manager for TLS
    cert_manager: Option<Arc<CertManager>>,
    /// Shared SNI certificate resolver used by EVERY live HTTPS listener this
    /// manager starts (`listen_on_tls` + the `:443` ingress).
    ///
    /// Created once in [`ProxyManager::new`] and cloned into each listener so a
    /// certificate hot-loaded into it (by the ACME provisioning trigger in
    /// [`Self::add_service`] or the daemon's renewal task) takes effect on the
    /// LIVE listener immediately — no restart. Before this was shared, each
    /// listener built its own resolver and dropped the handle, so a freshly
    /// provisioned cert could never reach the running listener and public
    /// HTTPS vhosts served "No certificate found" forever.
    sni_resolver: Arc<SniCertResolver>,
    /// Set of hosts for which certificate provisioning has already been
    /// scheduled, so a reconcile that re-runs [`Self::add_service`] for the same
    /// vhost does not re-spawn an ACME attempt every time. `CertManager::get_cert`
    /// is already cheap on a cache/disk hit, but a host whose provisioning is
    /// in-flight or has FAILED would otherwise be retried on every reconcile and
    /// burn Let's Encrypt rate limit; this guard makes the trigger fire at most
    /// once per host for the manager's lifetime.
    provisioning_requested: RwLock<HashSet<String>>,
    /// Ports with active TCP stream listeners (to avoid double-binding)
    tcp_listeners: RwLock<HashSet<u16>>,
    /// Ports with active UDP stream listeners (to avoid double-binding)
    udp_listeners: RwLock<HashSet<u16>>,
    /// Number of active proxy connections (for graceful drain on shutdown)
    active_connections: Arc<AtomicU64>,
    /// Optional network policy checker for access control enforcement
    network_policy_checker: Option<NetworkPolicyChecker>,
    /// Dedicated stream registry for node-loopback (`127.0.0.1:<port>`)
    /// publishing.
    ///
    /// This is intentionally separate from [`Self::stream_registry`]: the
    /// latter is keyed by endpoint port and entangled with the L7/L4 +
    /// Public/Internal binding matrix (`ensure_ports_for_service`). The
    /// loopback path forwards the node's `127.0.0.1:<endpoint.port>` to the
    /// container's real backend, independent of how the endpoint is exposed,
    /// so it owns its own registry and listener set.
    loopback_registry: Arc<StreamRegistry>,
    /// Active loopback TCP listeners keyed by published port. The
    /// [`JoinHandle`] owns the bound socket via its accept loop; aborting it
    /// frees the OS port. Used for both dedup and cleanup.
    loopback_tcp: RwLock<HashMap<u16, tokio::task::JoinHandle<()>>>,
    /// Active loopback UDP listeners keyed by published port. See
    /// [`Self::loopback_tcp`].
    loopback_udp: RwLock<HashMap<u16, tokio::task::JoinHandle<()>>>,
    /// Dedicated stream registry for explicit `port_mappings` host-port
    /// publishing (Docker-style `host_port -> container_port`). Separate from
    /// `loopback_registry`: the loopback path always binds `127.0.0.1:<port>`
    /// and forwards to the endpoint's target port; this path binds the
    /// mapping's `host_ip:host_port` (default `0.0.0.0`) and forwards to the
    /// container's `container_port`, independent of any endpoint.
    port_map_registry: Arc<StreamRegistry>,
    /// Active `port_mappings` TCP listeners keyed by bound host port.
    port_map_tcp: RwLock<HashMap<u16, tokio::task::JoinHandle<()>>>,
    /// Active `port_mappings` UDP listeners keyed by bound host port.
    port_map_udp: RwLock<HashMap<u16, tokio::task::JoinHandle<()>>>,
    /// Ownership map for published host-port loopback bindings.
    ///
    /// A host port (`127.0.0.1:<port>`) is a GLOBAL host resource — it can be
    /// bound exactly once. This maps each published port to the
    /// `(deployment, service)` that owns it. The first publisher claims the
    /// port; a second publisher for the SAME `(deployment, service)` is a
    /// legitimate scale-up (replica backend appended); a publisher for a
    /// DIFFERENT `(deployment, service)` is REFUSED (it would otherwise be
    /// silently appended into the foreign pool, so `:<port>` would serve the
    /// wrong deployment's backends — Bug 7). The entry is freed when the
    /// owning service's last backend on the port is unpublished.
    ///
    /// The key is the published port; the value is
    /// `(deployment, service_name)` where `deployment` is `None` for
    /// standalone callers.
    published_ports: RwLock<HashMap<u16, (Option<String>, String)>>,
    /// Background TCP health-check task for the L7 load balancer. Periodically
    /// TCP-connects to every registered backend and flips its health status,
    /// so a backend that was marked unhealthy by a transient request-path
    /// failure (e.g. the overlay momentarily reconfiguring while sibling
    /// containers churn during a CI build) AUTO-RECOVERS once it answers
    /// connects again. Without this the L7 LB had no recovery path of its own
    /// and a single transient blip left a service stuck on "no healthy
    /// backends" until a daemon restart. Aborted on drop.
    lb_health_checker: tokio::task::JoinHandle<()>,
    /// Whether the standing `0.0.0.0:80` / `0.0.0.0:443` ingress listeners have
    /// already been started. Makes [`ProxyManager::start_ingress`] idempotent
    /// so a double call (or a restart path) does not spawn duplicate
    /// retry-bind tasks that would fight over the same ports.
    ingress_started: AtomicBool,
    /// Per-service request-rate counter populated by the proxy request path.
    /// Threaded into every [`ProxyServer`] this manager builds and exposed via
    /// [`ProxyManager::rps_registry`] so the scheduler can read a real RPS
    /// signal (wrap it with [`RpsRegistryProvider`]).
    rps_registry: Arc<RpsRegistry>,
    /// Optional on-demand activator for scale-to-zero services. Installed by the
    /// daemon AFTER the `ServiceManager` exists (see
    /// [`ProxyManager::set_activator`]); threaded into every [`ProxyServer`]
    /// built thereafter. A `None` here preserves the existing immediate-`503`
    /// behavior.
    activator: RwLock<Option<Arc<dyn Activator>>>,
}

impl Drop for ProxyManager {
    fn drop(&mut self) {
        self.lb_health_checker.abort();
    }
}

impl ProxyManager {
    /// Create a new `ProxyManager` with the given configuration, service registry,
    /// and optional certificate manager.
    pub fn new(
        config: ProxyManagerConfig,
        registry: Arc<ServiceRegistry>,
        cert_manager: Option<Arc<CertManager>>,
    ) -> Self {
        let load_balancer = Arc::new(LoadBalancer::new());

        // Spawn the L7 load balancer's own TCP health checker so unhealthy
        // backends auto-recover. Probe every 5s with a 2s per-probe timeout:
        // fast enough that a transient blip during a CI build (sibling
        // containers churning the overlay) clears well within a single e2e
        // step, without hammering backends.
        let lb_health_checker =
            load_balancer.spawn_health_checker(Duration::from_secs(5), Duration::from_secs(2));

        Self {
            config,
            registry,
            load_balancer,
            servers: RwLock::new(HashMap::new()),
            services: RwLock::new(HashMap::new()),
            stream_registry: None,
            cert_manager,
            sni_resolver: Arc::new(SniCertResolver::new()),
            provisioning_requested: RwLock::new(HashSet::new()),
            tcp_listeners: RwLock::new(HashSet::new()),
            udp_listeners: RwLock::new(HashSet::new()),
            active_connections: Arc::new(AtomicU64::new(0)),
            network_policy_checker: None,
            loopback_registry: Arc::new(StreamRegistry::new()),
            loopback_tcp: RwLock::new(HashMap::new()),
            loopback_udp: RwLock::new(HashMap::new()),
            port_map_registry: Arc::new(StreamRegistry::new()),
            port_map_tcp: RwLock::new(HashMap::new()),
            port_map_udp: RwLock::new(HashMap::new()),
            published_ports: RwLock::new(HashMap::new()),
            lb_health_checker,
            ingress_started: AtomicBool::new(false),
            rps_registry: Arc::new(RpsRegistry::new()),
            activator: RwLock::new(None),
        }
    }

    /// Get a reference to the service registry
    pub fn registry(&self) -> Arc<ServiceRegistry> {
        self.registry.clone()
    }

    /// Get a reference to the load balancer
    pub fn load_balancer(&self) -> Arc<LoadBalancer> {
        self.load_balancer.clone()
    }

    /// Get the shared per-service [`RpsRegistry`].
    ///
    /// The proxy request path records every routed request into this registry;
    /// wrap it with [`RpsRegistryProvider`] to hand the scheduler an
    /// [`RpsProvider`] backed by real proxy traffic.
    #[must_use]
    pub fn rps_registry(&self) -> Arc<RpsRegistry> {
        Arc::clone(&self.rps_registry)
    }

    /// Get the shared [`RpsRegistry`] adapted as a scheduler [`RpsProvider`].
    ///
    /// Convenience wrapper around [`Self::rps_registry`] for the scaling
    /// wiring: the returned value implements
    /// [`zlayer_scheduler::scalers::RpsProvider`].
    #[must_use]
    pub fn rps_provider(&self) -> RpsRegistryProvider {
        RpsRegistryProvider::new(self.rps_registry())
    }

    /// Install the on-demand [`Activator`] used for scale-to-zero wake-ups.
    ///
    /// Call this once the agent's `ServiceManager` exists: build a
    /// [`ServiceActivator`] from a [`ScaleTrigger`] (the `ServiceManager`) plus
    /// [`Self::load_balancer`], then pass it here. Servers built AFTER this call
    /// (including [`Self::start_ingress`]) carry the activator; a request to an
    /// idle service then triggers a wake-up instead of an immediate `503`.
    pub async fn set_activator(&self, activator: Arc<dyn Activator>) {
        *self.activator.write().await = Some(activator);
    }

    /// Build a [`ServiceActivator`] from `trigger` and this manager's load
    /// balancer, then install it via [`Self::set_activator`].
    ///
    /// Convenience for the common wiring where the agent's `ServiceManager`
    /// implements [`ScaleTrigger`].
    pub async fn install_service_activator(&self, trigger: Arc<dyn ScaleTrigger>) {
        let activator = Arc::new(ServiceActivator::new(trigger, self.load_balancer.clone()));
        self.set_activator(activator).await;
    }

    /// Snapshot the currently-installed activator, if any. Used internally when
    /// building servers so each server carries the activator present at build
    /// time.
    async fn current_activator(&self) -> Option<Arc<dyn Activator>> {
        self.activator.read().await.clone()
    }

    /// Get the number of currently active proxy connections.
    pub fn active_connections(&self) -> u64 {
        self.active_connections.load(Ordering::Relaxed)
    }

    /// Get a reference to the certificate manager (if configured)
    pub fn cert_manager(&self) -> Option<&Arc<CertManager>> {
        self.cert_manager.as_ref()
    }

    /// Get the shared [`SniCertResolver`] backing every live HTTPS listener.
    ///
    /// Hand this to the daemon's certificate renewal task so renewed certs are
    /// hot-loaded into the SAME resolver the running `:443` listener serves
    /// from — `refresh_cert` on this handle is visible to live connections
    /// immediately, with no listener restart.
    #[must_use]
    pub fn sni_resolver(&self) -> Arc<SniCertResolver> {
        Arc::clone(&self.sni_resolver)
    }

    /// Set the stream registry for L4 proxy integration (TCP/UDP)
    pub fn set_stream_registry(&mut self, registry: Arc<StreamRegistry>) {
        self.stream_registry = Some(registry);
    }

    /// Builder pattern: add stream registry for L4 proxy integration
    #[must_use]
    pub fn with_stream_registry(mut self, registry: Arc<StreamRegistry>) -> Self {
        self.stream_registry = Some(registry);
        self
    }

    /// Get the stream registry (if configured)
    pub fn stream_registry(&self) -> Option<&Arc<StreamRegistry>> {
        self.stream_registry.as_ref()
    }

    /// Set the network policy checker for access control enforcement
    pub fn set_network_policy_checker(&mut self, checker: NetworkPolicyChecker) {
        self.network_policy_checker = Some(checker);
    }

    /// Builder pattern: add network policy checker for access control enforcement
    #[must_use]
    pub fn with_network_policy_checker(mut self, checker: NetworkPolicyChecker) -> Self {
        self.network_policy_checker = Some(checker);
        self
    }

    /// Start listening on a specific port bound to the given address.
    ///
    /// If already listening on this port, skip.
    /// All port listeners share the same `ServiceRegistry` for request matching.
    ///
    /// # Errors
    /// Returns an error if the proxy server cannot be started.
    pub async fn listen_on(&self, port: u16, bind_ip: IpAddr) -> Result<()> {
        let mut servers = self.servers.write().await;

        if servers.contains_key(&port) {
            debug!(port = port, "Already listening on port");
            return Ok(());
        }

        let addr = SocketAddr::new(bind_ip, port);
        let mut proxy_config = ProxyConfig::default();
        proxy_config.server.http_addr = addr;
        proxy_config.server.http2_enabled = self.config.http2_enabled;

        let mut server = ProxyServer::with_registry(
            proxy_config,
            self.registry.clone(),
            self.load_balancer.clone(),
        )
        .with_rps_registry(self.rps_registry());
        if let Some(ref checker) = self.network_policy_checker {
            server = server.with_network_policy_checker(checker.clone());
        }
        if let Some(activator) = self.current_activator().await {
            server = server.with_activator(activator);
        }
        let server = Arc::new(server);

        info!(port = port, bind = %addr, "Proxy listening on port");

        let server_clone = server.clone();
        tokio::spawn(async move {
            if let Err(e) = server_clone.run().await {
                tracing::error!(port = port, error = %e, "Proxy server error on port");
            }
        });

        servers.insert(port, server);
        Ok(())
    }

    /// Start an HTTPS listener on the given port using `SniCertResolver` for dynamic cert selection.
    ///
    /// If already listening on this port, skip.
    /// Requires a `CertManager` to be configured; logs a warning and returns `Ok(())` if not.
    ///
    /// # Errors
    /// Returns an error if the HTTPS proxy server cannot be started.
    pub async fn listen_on_tls(&self, port: u16, bind_ip: IpAddr) -> Result<()> {
        let mut servers = self.servers.write().await;

        if servers.contains_key(&port) {
            debug!(port = port, "Already listening on port (TLS)");
            return Ok(());
        }

        let Some(cert_manager) = &self.cert_manager else {
            warn!(
                port = port,
                "Cannot start TLS listener: no CertManager configured"
            );
            return Ok(());
        };

        // Use the manager's SHARED SniCertResolver so certs hot-loaded later
        // (ACME provisioning trigger / renewal task) reach this live listener.
        let sni_resolver = self.sni_resolver();

        // Load existing certificates (best-effort; log warnings on failure)
        let _ = load_existing_certs_into_resolver(cert_manager, &sni_resolver).await;

        let addr = SocketAddr::new(bind_ip, port);
        let mut proxy_config = ProxyConfig::default();
        proxy_config.server.https_addr = addr;

        let mut server = ProxyServer::with_tls_resolver(
            proxy_config,
            self.registry.clone(),
            self.load_balancer.clone(),
            sni_resolver,
        )
        .with_cert_manager(Arc::clone(cert_manager))
        .with_rps_registry(self.rps_registry());
        if let Some(ref checker) = self.network_policy_checker {
            server = server.with_network_policy_checker(checker.clone());
        }
        if let Some(activator) = self.current_activator().await {
            server = server.with_activator(activator);
        }
        let server = Arc::new(server);

        info!(port = port, bind = %addr, "HTTPS proxy listening on port");

        let server_clone = server.clone();
        tokio::spawn(async move {
            if let Err(e) = server_clone.run_https().await {
                tracing::error!(port = port, error = %e, "HTTPS proxy server error");
            }
        });

        servers.insert(port, server);
        Ok(())
    }

    /// Start the standing HTTP/HTTPS ingress on `0.0.0.0:80` and `0.0.0.0:443`.
    ///
    /// The lone daemon IS the ingress: this binds the two well-known web ports
    /// so every deployed service with an HTTP/HTTPS endpoint is reachable on
    /// the standard ports, routed by the shared [`ServiceRegistry`]/SNI cert
    /// resolver the manager already holds — no per-service listener config.
    ///
    /// **Conflict policy: WARN, never error.** If 80/443 is already held the
    /// underlying [`ProxyServer::run_with_retry`] /
    /// [`ProxyServer::run_https_with_retry`] log a warning and keep retrying
    /// the bind forever, grabbing the port the moment it frees. This NEVER
    /// aborts startup, NEVER blocks deployments, and NEVER hard-errors. Binding
    /// 80/443 needs root or `CAP_NET_BIND_SERVICE`; a non-root daemon simply
    /// never grabs them and keeps warning — that is fine.
    ///
    /// Idempotent: a second call is a no-op (it will not spawn duplicate
    /// retry-bind tasks).
    pub async fn start_ingress(&self) {
        self.start_ingress_on(DEFAULT_INGRESS_HTTP_PORT, DEFAULT_INGRESS_HTTPS_PORT)
            .await;
    }

    /// Build (but do not start) the `:80` HTTP ingress [`ProxyServer`].
    ///
    /// Applies the shared registry, RPS registry, network-policy checker, the
    /// supplied on-demand `activator`, AND — crucially — this manager's
    /// [`CertManager`] when one is configured. The cert manager on `:80` exists
    /// for exactly one reason: ACME HTTP-01 challenges
    /// (`/.well-known/acme-challenge/<token>`) ALWAYS arrive on `:80`, and the
    /// request path serves them only when `cert_manager` is `Some`. There is no
    /// TLS on `:80`; this is purely the HTTP-01 carve-out.
    ///
    /// Network-free: constructs the server without binding any port, so callers
    /// (and tests) can inspect the result before [`Self::start_ingress_on`]
    /// spawns the retry-bind task.
    fn build_http_ingress_server(
        &self,
        http_addr: SocketAddr,
        activator: Option<Arc<dyn Activator>>,
    ) -> ProxyServer {
        let mut http_proxy_config = ProxyConfig::default();
        http_proxy_config.server.http_addr = http_addr;
        http_proxy_config.server.http2_enabled = self.config.http2_enabled;

        let mut http_server = ProxyServer::with_registry(
            http_proxy_config,
            self.registry.clone(),
            self.load_balancer.clone(),
        )
        .with_rps_registry(self.rps_registry());
        if let Some(ref checker) = self.network_policy_checker {
            http_server = http_server.with_network_policy_checker(checker.clone());
        }
        if let Some(activator) = activator {
            http_server = http_server.with_activator(activator);
        }
        // Wire the cert manager so the `:80` listener can serve ACME HTTP-01
        // challenges. Without this the interception is skipped and the request
        // falls through to the vhost match, which 403s when no HTTP service is
        // registered for the challenge domain — so certs can never be issued.
        if let Some(ref cm) = self.cert_manager {
            http_server = http_server.with_cert_manager(Arc::clone(cm));
        }
        http_server
    }

    /// Like [`Self::start_ingress`] but with explicit HTTP/HTTPS ports.
    ///
    /// Used by tests; the production path uses [`DEFAULT_INGRESS_HTTP_PORT`] /
    /// [`DEFAULT_INGRESS_HTTPS_PORT`].
    #[allow(clippy::similar_names)]
    pub async fn start_ingress_on(&self, http_port: u16, https_port: u16) {
        // Idempotency: only the first caller wins.
        if self
            .ingress_started
            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
            .is_err()
        {
            debug!("Ingress already started; skipping");
            return;
        }

        let bind_ip = IpAddr::V4(Ipv4Addr::UNSPECIFIED); // 0.0.0.0

        // ---- HTTP ingress (:80) ----
        let http_addr = SocketAddr::new(bind_ip, http_port);
        let activator = self.current_activator().await;
        let http_server = Arc::new(self.build_http_ingress_server(http_addr, activator));
        info!(port = http_port, bind = %http_addr, "Starting HTTP ingress (retry-never-error)");
        {
            let server = http_server.clone();
            tokio::spawn(async move {
                if let Err(e) = server.run_with_retry(http_addr).await {
                    // Only reached on a post-bind fatal accept-loop error; the
                    // bind itself never errors out.
                    warn!(port = http_port, error = %e, "HTTP ingress accept loop exited");
                }
            });
        }
        // Additional v6-only listener on `[::]:<http_port>` so IPv6 clients can
        // reach hosted vhosts. Shares the same `http_server` Arc (and thus its
        // shutdown signal); a missing-IPv6 host stays non-fatal.
        {
            let server = http_server.clone();
            tokio::spawn(async move {
                match bind_v6_only(http_port) {
                    Ok(l) => {
                        if let Err(e) = server.run_on_listener(l).await {
                            warn!(port = http_port, error = %e, "HTTP v6 ingress exited");
                        }
                    }
                    Err(e) => {
                        warn!(port = http_port, error = %e, "HTTP v6 ingress bind failed (non-fatal)");
                    }
                }
            });
        }
        self.servers.write().await.insert(http_port, http_server);

        // ---- HTTPS ingress (:443) ----
        let Some(cert_manager) = &self.cert_manager else {
            warn!(
                port = https_port,
                "Cannot start HTTPS ingress: no CertManager configured (HTTP ingress is up)"
            );
            return;
        };

        // Use the manager's SHARED SniCertResolver so certs hot-loaded later
        // (ACME provisioning trigger / renewal task) reach this live listener.
        let sni_resolver = self.sni_resolver();
        // Load existing certificates (best-effort; log warnings on failure).
        let _ = load_existing_certs_into_resolver(cert_manager, &sni_resolver).await;

        let https_addr = SocketAddr::new(bind_ip, https_port);
        let mut https_proxy_config = ProxyConfig::default();
        https_proxy_config.server.https_addr = https_addr;

        let mut https_server = ProxyServer::with_tls_resolver(
            https_proxy_config,
            self.registry.clone(),
            self.load_balancer.clone(),
            sni_resolver,
        )
        .with_cert_manager(Arc::clone(cert_manager))
        .with_rps_registry(self.rps_registry());
        if let Some(ref checker) = self.network_policy_checker {
            https_server = https_server.with_network_policy_checker(checker.clone());
        }
        if let Some(activator) = self.current_activator().await {
            https_server = https_server.with_activator(activator);
        }
        let https_server = Arc::new(https_server);
        info!(port = https_port, bind = %https_addr, "Starting HTTPS ingress (retry-never-error)");
        {
            let server = https_server.clone();
            tokio::spawn(async move {
                if let Err(e) = server.run_https_with_retry(https_addr).await {
                    warn!(port = https_port, error = %e, "HTTPS ingress accept loop exited");
                }
            });
        }
        // Additional v6-only TLS listener on `[::]:<https_port>`. Reuses the
        // server's configured TlsAcceptor (same SNI resolver) via
        // `run_https_on_listener`; shares the `https_server` Arc's shutdown.
        {
            let server = https_server.clone();
            tokio::spawn(async move {
                match bind_v6_only(https_port) {
                    Ok(l) => {
                        if let Err(e) = server.run_https_on_listener(l).await {
                            warn!(port = https_port, error = %e, "HTTPS v6 ingress exited");
                        }
                    }
                    Err(e) => {
                        warn!(port = https_port, error = %e, "HTTPS v6 ingress bind failed (non-fatal)");
                    }
                }
            });
        }
        self.servers.write().await.insert(https_port, https_server);
    }

    /// Stop all proxy servers on all ports.
    ///
    /// After signalling each server to shut down, waits up to 30 seconds for
    /// active connections to drain before returning.
    pub async fn stop(&self) {
        let mut servers = self.servers.write().await;
        for (port, server) in servers.drain() {
            info!(port = port, "Stopping proxy on port");
            server.shutdown();
        }

        // Wait up to 30s for active connections to drain
        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
        while self.active_connections.load(Ordering::Relaxed) > 0 {
            if tokio::time::Instant::now() >= deadline {
                let remaining = self.active_connections.load(Ordering::Relaxed);
                warn!(
                    remaining = remaining,
                    "Drain timeout reached, forcing shutdown"
                );
                break;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        info!("All proxy servers stopped");
    }

    /// Remove and shut down the listener on a specific port.
    pub async fn unbind(&self, port: u16) {
        let mut servers = self.servers.write().await;
        if let Some(server) = servers.remove(&port) {
            info!(port = port, "Unbinding proxy from port");
            server.shutdown();
        }
    }

    /// Scan a service's endpoints and ensure the proxy is listening on all
    /// required ports.
    ///
    /// - **HTTP/HTTPS/WebSocket** endpoints start an HTTP proxy listener.
    /// - **TCP** endpoints bind a `TcpListener` and spawn a `TcpStreamService`.
    /// - **UDP** endpoints bind a `UdpSocket` and spawn a `UdpStreamService`.
    ///
    /// Bind address is determined by the `expose` type:
    /// - **Public** endpoints bind to `0.0.0.0` (all interfaces).
    /// - **Internal** endpoints bind to the overlay IP so they are only
    ///   reachable from within the overlay network.  If no overlay is
    ///   available, internal endpoints bind to `127.0.0.1` (localhost only).
    ///
    /// # Errors
    /// Returns an error if an HTTP/HTTPS listener cannot be started.
    pub async fn ensure_ports_for_service(
        &self,
        spec: &ServiceSpec,
        overlay_ip: Option<IpAddr>,
    ) -> Result<()> {
        for endpoint in &spec.endpoints {
            let bind_ip = match endpoint.expose {
                ExposeType::Public => IpAddr::V4(Ipv4Addr::UNSPECIFIED), // 0.0.0.0
                ExposeType::Internal => {
                    // Prefer overlay IP; fall back to loopback if overlay is unavailable.
                    let ip = overlay_ip.unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST));
                    if overlay_ip.is_none() {
                        warn!(
                            endpoint = %endpoint.name,
                            port = endpoint.port,
                            "No overlay IP available for internal endpoint; binding to 127.0.0.1"
                        );
                    }
                    ip
                }
            };

            match endpoint.protocol {
                Protocol::Https => {
                    // L7 TLS: start HTTPS proxy listener with SNI cert resolution
                    self.listen_on_tls(endpoint.port, bind_ip).await?;
                }
                Protocol::Http | Protocol::Websocket => {
                    // L7: start HTTP proxy listener
                    self.listen_on(endpoint.port, bind_ip).await?;
                }
                Protocol::Tcp => {
                    // L4 TCP: bind listener and spawn TcpStreamService.
                    // Translate the endpoint's `stream:` block into the proxy's
                    // runtime config (TLS / PROXY-protocol / health-check).
                    let cfg = translate_stream_config(endpoint.stream.as_ref());
                    self.ensure_tcp_listener(endpoint.port, bind_ip, &cfg).await;
                }
                Protocol::Udp => {
                    // L4 UDP: bind socket and spawn UdpStreamService.
                    let cfg = translate_stream_config(endpoint.stream.as_ref());
                    self.ensure_udp_listener(endpoint.port, bind_ip, &cfg).await;
                }
            }
        }
        Ok(())
    }

    /// Ensure a TCP stream listener is running on the given port.
    ///
    /// If a listener is already active on this port, this is a no-op.
    /// Requires `stream_registry` to be configured; logs a warning if not.
    ///
    /// `cfg` carries the endpoint's translated `stream:` settings: when
    /// `cfg.tls` is set and a [`CertManager`] is configured, the listener
    /// terminates TLS using the shared SNI cert resolver; `cfg.proxy_protocol`
    /// makes the listener prepend a PROXY v2 header to upstream connections;
    /// and `cfg.health_check` is applied to the registered service so the
    /// background health checker probes it.
    async fn ensure_tcp_listener(&self, port: u16, bind_ip: IpAddr, cfg: &StreamProxyConfig) {
        // Apply the L4 health-check config to the service registered for this
        // port (registration of backends happens out-of-band in ServiceManager).
        self.apply_stream_health_check_tcp(port, cfg);

        // Check if already listening
        {
            let listeners = self.tcp_listeners.read().await;
            if listeners.contains(&port) {
                debug!(port = port, "TCP stream listener already active");
                return;
            }
        }

        let registry = if let Some(r) = &self.stream_registry {
            Arc::clone(r)
        } else {
            warn!(
                port = port,
                "Cannot start TCP listener: StreamRegistry not configured"
            );
            return;
        };

        let addr = SocketAddr::new(bind_ip, port);
        let listener = match tokio::net::TcpListener::bind(addr).await {
            Ok(l) => l,
            Err(e) => {
                warn!(
                    port = port,
                    bind = %addr,
                    error = %e,
                    "Failed to bind TCP stream listener, continuing"
                );
                return;
            }
        };

        // Mark as active before spawning
        {
            let mut listeners = self.tcp_listeners.write().await;
            listeners.insert(port);
        }

        let mut tcp_service = TcpStreamService::new(registry, port);
        if cfg.proxy_protocol {
            tcp_service = tcp_service.with_proxy_protocol(true);
        }
        // L4 TLS termination: reuse the shared SNI cert resolver (same certs as
        // HTTPS endpoints). Only enabled when a CertManager backs the resolver.
        if cfg.tls {
            if self.cert_manager.is_some() {
                let resolver = self.sni_resolver();
                if let Some(cert_manager) = &self.cert_manager {
                    let _ = load_existing_certs_into_resolver(cert_manager, &resolver).await;
                }
                let acceptor = tls_acceptor_from_resolver(resolver);
                tcp_service = tcp_service.with_tls_acceptor(acceptor);
                info!(port = port, "L4 TCP TLS termination enabled");
            } else {
                warn!(
                    port = port,
                    "stream.tls requested but no CertManager configured; serving TCP without TLS"
                );
            }
        }

        let tcp_service = Arc::new(tcp_service);
        tokio::spawn(async move {
            tcp_service.serve(listener).await;
        });

        info!(port = port, bind = %addr, "TCP stream proxy listening");
    }

    /// Apply a translated L4 config (incl. the health probe) to the TCP service
    /// registered on `port` (if any). No-op when no `stream_registry` / service
    /// is present.
    fn apply_stream_health_check_tcp(&self, port: u16, cfg: &StreamProxyConfig) {
        if let Some(registry) = &self.stream_registry {
            registry.set_tcp_config(port, cfg.clone());
        }
    }

    /// Apply a translated L4 config (incl. the health probe) to the UDP service
    /// registered on `port` (if any). No-op when no `stream_registry` / service
    /// is present.
    fn apply_stream_health_check_udp(&self, port: u16, cfg: &StreamProxyConfig) {
        if let Some(registry) = &self.stream_registry {
            registry.set_udp_config(port, cfg.clone());
        }
    }

    /// Ensure a UDP stream listener is running on the given port.
    ///
    /// If a listener is already active on this port, this is a no-op.
    /// Requires `stream_registry` to be configured; logs a warning if not.
    ///
    /// `cfg.session_timeout` overrides the per-session UDP idle timeout, and
    /// `cfg.health_check` is applied to the registered service so the
    /// background health checker performs the configured UDP probe.
    async fn ensure_udp_listener(&self, port: u16, bind_ip: IpAddr, cfg: &StreamProxyConfig) {
        // Apply the L4 health-check config to the service registered for this
        // port (backend registration happens out-of-band in ServiceManager).
        self.apply_stream_health_check_udp(port, cfg);

        // Check if already listening
        {
            let listeners = self.udp_listeners.read().await;
            if listeners.contains(&port) {
                debug!(port = port, "UDP stream listener already active");
                return;
            }
        }

        let registry = if let Some(r) = &self.stream_registry {
            Arc::clone(r)
        } else {
            warn!(
                port = port,
                "Cannot start UDP listener: StreamRegistry not configured"
            );
            return;
        };

        let addr = SocketAddr::new(bind_ip, port);
        let socket = match tokio::net::UdpSocket::bind(addr).await {
            Ok(s) => s,
            Err(e) => {
                warn!(
                    port = port,
                    bind = %addr,
                    error = %e,
                    "Failed to bind UDP stream listener, continuing"
                );
                return;
            }
        };

        // Mark as active before spawning
        {
            let mut listeners = self.udp_listeners.write().await;
            listeners.insert(port);
        }

        let udp_service = Arc::new(UdpStreamService::new(registry, port, cfg.session_timeout));
        tokio::spawn(async move {
            if let Err(e) = udp_service.serve(socket).await {
                tracing::error!(
                    port = port,
                    error = %e,
                    "UDP stream proxy service failed"
                );
            }
        });

        info!(port = port, bind = %addr, "UDP stream proxy listening");
    }

    /// Publish a single container's exposed ports on the node loopback
    /// (`127.0.0.1:<endpoint.port>`), forwarding to wherever the container
    /// actually listens.
    ///
    /// This implements the GitHub-Actions "service published to localhost"
    /// convention so a consumer sharing the node loopback can reach the
    /// service at `localhost:<port>`. The published port is always
    /// `endpoint.port`; the backend the listener forwards to is
    /// `(container_ip, port_override.unwrap_or(endpoint.target_port()))`,
    /// which is already runtime-resolved by the caller:
    ///
    /// - On the macOS seatbelt/libkrun runtimes every replica shares the host
    ///   `127.0.0.1` and gets a unique `port_override`, so the container
    ///   listens on `127.0.0.1:<port_override>` and we forward there.
    /// - On Linux/VZ/HCS the container listens on its overlay IP, so
    ///   `container_ip` is the overlay address and `port_override` is `None`,
    ///   forwarding to `overlay_ip:<target_port>`.
    ///
    /// Backends accumulate across replicas so multiple members round-robin
    /// behind the single loopback port. `Public` endpoints are skipped: they
    /// are already bound on `0.0.0.0` and therefore already reachable on
    /// loopback — binding `127.0.0.1:<port>` again would fail with
    /// `EADDRINUSE`.
    ///
    /// This NEVER rewrites a container's own loopback: it only binds the
    /// NODE's `127.0.0.1` and forwards to the container's runtime-resolved
    /// address.
    ///
    /// Bind failures are tolerated (logged at `warn!`); this never panics on
    /// them.
    ///
    /// A published host port (`127.0.0.1:<port>`) is a GLOBAL host resource and
    /// is OWNED by the first `(deployment, service)` to publish it. A second
    /// publish for the SAME `(deployment, service)` appends a replica backend
    /// (legitimate scale-up). A publish for a DIFFERENT `(deployment, service)`
    /// is REFUSED with [`AgentError::PortConflict`] rather than silently
    /// appended into the foreign pool (Bug 7: that would make `:<port>` serve
    /// the wrong deployment's backends). On a conflict for any endpoint this
    /// returns `Err` after having published the conflict-free endpoints up to
    /// that point.
    ///
    /// `deployment` is the owning deployment name (`Some`) or `None` for
    /// standalone callers.
    ///
    /// # Errors
    /// Returns [`AgentError::PortConflict`] when a published port is already
    /// owned by a different `(deployment, service)`.
    pub async fn publish_loopback_for_container(
        &self,
        deployment: Option<&str>,
        service_name: &str,
        spec: &ServiceSpec,
        container_ip: IpAddr,
        port_override: Option<u16>,
    ) -> Result<()> {
        for endpoint in &spec.endpoints {
            // Public endpoints already bind 0.0.0.0 -> already on loopback.
            if matches!(endpoint.expose, ExposeType::Public) {
                continue;
            }

            let backend = SocketAddr::new(
                container_ip,
                port_override.unwrap_or_else(|| endpoint.target_port()),
            );
            let publish_port = endpoint.port;

            // Enforce host-port ownership before touching any registry.
            self.claim_published_port(deployment, service_name, publish_port)
                .await?;

            match endpoint.protocol {
                Protocol::Tcp | Protocol::Http | Protocol::Https | Protocol::Websocket => {
                    // A raw TCP forward carries HTTP/HTTPS/WS just fine, so
                    // all L7 protocols ride the loopback TCP path.
                    self.publish_loopback_tcp(service_name, publish_port, backend)
                        .await;
                }
                Protocol::Udp => {
                    self.publish_loopback_udp(service_name, publish_port, backend)
                        .await;
                }
            }
        }
        Ok(())
    }

    /// Claim ownership of host port `publish_port` for `(deployment, service)`.
    ///
    /// - Unowned → claim it and return `Ok`.
    /// - Owned by the SAME `(deployment, service)` → return `Ok` (scale-up).
    /// - Owned by a DIFFERENT `(deployment, service)` → return
    ///   [`AgentError::PortConflict`] (refuse the cross-wire).
    async fn claim_published_port(
        &self,
        deployment: Option<&str>,
        service_name: &str,
        publish_port: u16,
    ) -> Result<()> {
        let mut owners = self.published_ports.write().await;
        if let Some((owner_dep, owner_svc)) = owners.get(&publish_port) {
            if owner_dep.as_deref() == deployment && owner_svc == service_name {
                // Same owner: legitimate scale-up / re-publish.
                return Ok(());
            }
            let owner = format!("{}/{}", owner_dep.as_deref().unwrap_or("_"), owner_svc);
            let requester = format!("{}/{}", deployment.unwrap_or("_"), service_name);
            warn!(
                port = publish_port,
                owner = %owner,
                requester = %requester,
                "Refusing to publish host port already owned by a different deployment/service (would cross-wire backends)"
            );
            return Err(crate::error::AgentError::PortConflict {
                port: publish_port,
                owner,
                requester,
            });
        }
        owners.insert(
            publish_port,
            (deployment.map(str::to_string), service_name.to_string()),
        );
        Ok(())
    }

    /// Register `backend` for the loopback TCP listener on `publish_port`,
    /// binding `127.0.0.1:<publish_port>` if it is not already bound.
    async fn publish_loopback_tcp(
        &self,
        service_name: &str,
        publish_port: u16,
        backend: SocketAddr,
    ) {
        // Accumulate the backend in the loopback registry.
        if let Some(existing) = self.loopback_registry.resolve_tcp(publish_port) {
            let mut backends = existing.backends;
            if !backends.contains(&backend) {
                backends.push(backend);
            }
            self.loopback_registry
                .update_tcp_backends(publish_port, backends);
        } else {
            self.loopback_registry.register_tcp(
                publish_port,
                StreamService::new(service_name.to_string(), vec![backend]),
            );
        }

        // Bind the loopback listener once per port.
        let mut listeners = self.loopback_tcp.write().await;
        if listeners.contains_key(&publish_port) {
            debug!(port = publish_port, "Loopback TCP listener already active");
            return;
        }

        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), publish_port);
        let listener = match tokio::net::TcpListener::bind(addr).await {
            Ok(l) => l,
            Err(e) => {
                warn!(
                    port = publish_port,
                    bind = %addr,
                    error = %e,
                    "Failed to bind loopback TCP listener, continuing"
                );
                return;
            }
        };

        let tcp_service = Arc::new(TcpStreamService::new(
            Arc::clone(&self.loopback_registry),
            publish_port,
        ));
        let handle = tokio::spawn(async move {
            tcp_service.serve(listener).await;
        });
        listeners.insert(publish_port, handle);
        drop(listeners);

        info!(
            service = service_name,
            port = publish_port,
            bind = %addr,
            backend = %backend,
            "Published service port on node loopback (TCP)"
        );
    }

    /// Register `backend` for the loopback UDP listener on `publish_port`,
    /// binding `127.0.0.1:<publish_port>` if it is not already bound.
    async fn publish_loopback_udp(
        &self,
        service_name: &str,
        publish_port: u16,
        backend: SocketAddr,
    ) {
        if let Some(existing) = self.loopback_registry.resolve_udp(publish_port) {
            let mut backends = existing.backends;
            if !backends.contains(&backend) {
                backends.push(backend);
            }
            self.loopback_registry
                .update_udp_backends(publish_port, backends);
        } else {
            self.loopback_registry.register_udp(
                publish_port,
                StreamService::new(service_name.to_string(), vec![backend]),
            );
        }

        let mut listeners = self.loopback_udp.write().await;
        if listeners.contains_key(&publish_port) {
            debug!(port = publish_port, "Loopback UDP listener already active");
            return;
        }

        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), publish_port);
        let socket = match tokio::net::UdpSocket::bind(addr).await {
            Ok(s) => s,
            Err(e) => {
                warn!(
                    port = publish_port,
                    bind = %addr,
                    error = %e,
                    "Failed to bind loopback UDP listener, continuing"
                );
                return;
            }
        };

        let udp_service = Arc::new(UdpStreamService::new(
            Arc::clone(&self.loopback_registry),
            publish_port,
            None,
        ));
        let handle = tokio::spawn(async move {
            if let Err(e) = udp_service.serve(socket).await {
                tracing::error!(
                    port = publish_port,
                    error = %e,
                    "Loopback UDP stream proxy service failed"
                );
            }
        });
        listeners.insert(publish_port, handle);
        drop(listeners);

        info!(
            service = service_name,
            port = publish_port,
            bind = %addr,
            backend = %backend,
            "Published service port on node loopback (UDP)"
        );
    }

    /// Remove a single container's backend from the node-loopback publish
    /// path. Mirrors [`Self::publish_loopback_for_container`]: it recomputes
    /// the same `(container_ip, port_override.unwrap_or(target_port))` backend
    /// per endpoint and drops it from the loopback registry.
    ///
    /// When a published port's backend set becomes empty, the registry entry
    /// is unregistered and the loopback listener is forgotten so the port is
    /// freed for the next bind. `Public` endpoints are skipped (they were
    /// never published here).
    pub async fn unpublish_loopback_for_container(
        &self,
        spec: &ServiceSpec,
        container_ip: IpAddr,
        port_override: Option<u16>,
    ) {
        for endpoint in &spec.endpoints {
            if matches!(endpoint.expose, ExposeType::Public) {
                continue;
            }

            let backend = SocketAddr::new(
                container_ip,
                port_override.unwrap_or_else(|| endpoint.target_port()),
            );
            let publish_port = endpoint.port;

            match endpoint.protocol {
                Protocol::Tcp | Protocol::Http | Protocol::Https | Protocol::Websocket => {
                    self.unpublish_loopback_tcp(publish_port, backend).await;
                }
                Protocol::Udp => {
                    self.unpublish_loopback_udp(publish_port, backend).await;
                }
            }
        }
    }

    /// Drop `backend` from the loopback TCP service on `publish_port`,
    /// freeing the listener when no backends remain.
    async fn unpublish_loopback_tcp(&self, publish_port: u16, backend: SocketAddr) {
        let Some(existing) = self.loopback_registry.resolve_tcp(publish_port) else {
            return;
        };
        let remaining: Vec<SocketAddr> = existing
            .backends
            .into_iter()
            .filter(|b| *b != backend)
            .collect();

        if remaining.is_empty() {
            let _ = self.loopback_registry.unregister_tcp(publish_port);
            let mut listeners = self.loopback_tcp.write().await;
            if let Some(handle) = listeners.remove(&publish_port) {
                handle.abort();
            }
            // Release host-port ownership so a different (deployment, service)
            // may bind it next.
            self.published_ports.write().await.remove(&publish_port);
            debug!(
                port = publish_port,
                "Freed loopback TCP listener (no backends remain)"
            );
        } else {
            self.loopback_registry
                .update_tcp_backends(publish_port, remaining);
        }
    }

    /// Drop `backend` from the loopback UDP service on `publish_port`,
    /// freeing the listener when no backends remain.
    async fn unpublish_loopback_udp(&self, publish_port: u16, backend: SocketAddr) {
        let Some(existing) = self.loopback_registry.resolve_udp(publish_port) else {
            return;
        };
        let remaining: Vec<SocketAddr> = existing
            .backends
            .into_iter()
            .filter(|b| *b != backend)
            .collect();

        if remaining.is_empty() {
            let _ = self.loopback_registry.unregister_udp(publish_port);
            let mut listeners = self.loopback_udp.write().await;
            if let Some(handle) = listeners.remove(&publish_port) {
                handle.abort();
            }
            // Release host-port ownership so a different (deployment, service)
            // may bind it next.
            self.published_ports.write().await.remove(&publish_port);
            debug!(
                port = publish_port,
                "Freed loopback UDP listener (no backends remain)"
            );
        } else {
            self.loopback_registry
                .update_udp_backends(publish_port, remaining);
        }
    }

    /// Publish one Docker-style port mapping (`host_ip:host_port -> container`)
    /// on the host, forwarding to the container's `container_port`.
    ///
    /// Unlike [`Self::publish_loopback_for_container`] this is driven by
    /// `spec.port_mappings`, NOT by endpoints, so a workload with port
    /// mappings and zero endpoints still gets a host listener. The listener
    /// binds `mapping.host_ip` (default `0.0.0.0`) on `mapping.host_port`; a
    /// `host_port` of `None` or `0` requests an OS-assigned ephemeral port.
    /// The backend forwarded to is `backend` (the caller resolves it to the
    /// container's runtime address + `container_port`/override).
    ///
    /// Returns the actually-bound host port (resolves ephemeral). Host-port
    /// ownership is enforced via the shared `published_ports` map: a second
    /// publish for the SAME `(deployment, owner)` appends a replica backend; a
    /// DIFFERENT owner on the same explicit port is refused with
    /// [`AgentError::PortConflict`].
    ///
    /// # Errors
    /// Returns [`AgentError::PortConflict`] on a cross-owner explicit-port
    /// collision, or an IO error if an ephemeral bind fails.
    pub async fn publish_port_mapping(
        &self,
        deployment: Option<&str>,
        owner_name: &str,
        mapping: &PortMapping,
        backend: SocketAddr,
    ) -> Result<u16> {
        let bind_ip: IpAddr = mapping
            .host_ip
            .parse()
            .unwrap_or(IpAddr::V4(Ipv4Addr::UNSPECIFIED));
        let requested = mapping.host_port.filter(|p| *p != 0);
        let is_udp = matches!(mapping.protocol, PortProtocol::Udp);
        let port = match mapping.protocol {
            PortProtocol::Udp => {
                self.publish_port_mapping_udp(deployment, owner_name, bind_ip, requested, backend)
                    .await?
            }
            // TCP (and any non-UDP) rides the TCP forward.
            PortProtocol::Tcp => {
                self.publish_port_mapping_tcp(deployment, owner_name, bind_ip, requested, backend)
                    .await?
            }
        };

        // Open the resolved host port on the host firewall so peers on other
        // nodes can reach a service published here — required for an
        // `OverlayMode::Shared` service (whose only ingress is the free-port
        // proxy) and correct for Docker-style published ports in general.
        // Idempotent (multiple replicas behind one host port re-call this),
        // best-effort + non-fatal.
        if let Err(e) = zlayer_overlay::firewall::ensure_published_port(port, is_udp) {
            warn!(
                error = %e,
                port,
                udp = is_udp,
                "could not open published port on host firewall (non-fatal)"
            );
        }
        Ok(port)
    }

    /// Bind `bind_ip:<port>` for a TCP port mapping, accumulating `backend`.
    /// `requested` is the explicit host port, or `None` for ephemeral.
    async fn publish_port_mapping_tcp(
        &self,
        deployment: Option<&str>,
        owner_name: &str,
        bind_ip: IpAddr,
        requested: Option<u16>,
        backend: SocketAddr,
    ) -> Result<u16> {
        if let Some(port) = requested {
            // Explicit port: claim ownership before binding.
            self.claim_published_port(deployment, owner_name, port)
                .await?;
            self.accumulate_port_map_tcp_backend(port, owner_name, backend);

            let mut listeners = self.port_map_tcp.write().await;
            if listeners.contains_key(&port) {
                debug!(port = port, "port-mapping TCP listener already active");
                return Ok(port);
            }
            let addr = SocketAddr::new(bind_ip, port);
            match tokio::net::TcpListener::bind(addr).await {
                Ok(listener) => {
                    let svc = Arc::new(TcpStreamService::new(
                        Arc::clone(&self.port_map_registry),
                        port,
                    ));
                    let handle = tokio::spawn(async move {
                        svc.serve(listener).await;
                    });
                    listeners.insert(port, handle);
                    drop(listeners);
                    info!(service = owner_name, port = port, bind = %addr, backend = %backend,
                        "Published port mapping (TCP)");
                    Ok(port)
                }
                Err(e) => {
                    warn!(port = port, bind = %addr, error = %e,
                        "Failed to bind port-mapping TCP listener, continuing");
                    Ok(port)
                }
            }
        } else {
            // Ephemeral: bind first to learn the port, then claim it.
            let addr = SocketAddr::new(bind_ip, 0);
            let listener = tokio::net::TcpListener::bind(addr).await.map_err(|e| {
                crate::error::AgentError::Network(format!(
                    "failed to bind ephemeral port-mapping TCP listener on {bind_ip}: {e}"
                ))
            })?;
            let port = listener.local_addr().map(|a| a.port()).map_err(|e| {
                crate::error::AgentError::Network(format!(
                    "failed to read ephemeral port-mapping TCP local addr: {e}"
                ))
            })?;
            self.claim_published_port(deployment, owner_name, port)
                .await?;
            self.accumulate_port_map_tcp_backend(port, owner_name, backend);
            let svc = Arc::new(TcpStreamService::new(
                Arc::clone(&self.port_map_registry),
                port,
            ));
            let handle = tokio::spawn(async move {
                svc.serve(listener).await;
            });
            self.port_map_tcp.write().await.insert(port, handle);
            info!(service = owner_name, port = port, bind = %bind_ip, backend = %backend,
                "Published port mapping on ephemeral host port (TCP)");
            Ok(port)
        }
    }

    /// Accumulate `backend` into the port-map TCP registry under `port`.
    fn accumulate_port_map_tcp_backend(&self, port: u16, owner_name: &str, backend: SocketAddr) {
        if let Some(existing) = self.port_map_registry.resolve_tcp(port) {
            let mut backends = existing.backends;
            if !backends.contains(&backend) {
                backends.push(backend);
            }
            self.port_map_registry.update_tcp_backends(port, backends);
        } else {
            self.port_map_registry.register_tcp(
                port,
                StreamService::new(owner_name.to_string(), vec![backend]),
            );
        }
    }

    /// Bind `bind_ip:<port>` for a UDP port mapping, accumulating `backend`.
    /// `requested` is the explicit host port, or `None` for ephemeral.
    async fn publish_port_mapping_udp(
        &self,
        deployment: Option<&str>,
        owner_name: &str,
        bind_ip: IpAddr,
        requested: Option<u16>,
        backend: SocketAddr,
    ) -> Result<u16> {
        if let Some(port) = requested {
            // Explicit port: claim ownership before binding.
            self.claim_published_port(deployment, owner_name, port)
                .await?;
            self.accumulate_port_map_udp_backend(port, owner_name, backend);

            let mut listeners = self.port_map_udp.write().await;
            if listeners.contains_key(&port) {
                debug!(port = port, "port-mapping UDP listener already active");
                return Ok(port);
            }
            let addr = SocketAddr::new(bind_ip, port);
            match tokio::net::UdpSocket::bind(addr).await {
                Ok(socket) => {
                    let svc = Arc::new(UdpStreamService::new(
                        Arc::clone(&self.port_map_registry),
                        port,
                        None,
                    ));
                    let handle = tokio::spawn(async move {
                        if let Err(e) = svc.serve(socket).await {
                            tracing::error!(
                                port = port,
                                error = %e,
                                "port-mapping UDP stream proxy service failed"
                            );
                        }
                    });
                    listeners.insert(port, handle);
                    drop(listeners);
                    info!(service = owner_name, port = port, bind = %addr, backend = %backend,
                        "Published port mapping (UDP)");
                    Ok(port)
                }
                Err(e) => {
                    warn!(port = port, bind = %addr, error = %e,
                        "Failed to bind port-mapping UDP listener, continuing");
                    Ok(port)
                }
            }
        } else {
            // Ephemeral: bind first to learn the port, then claim it.
            let addr = SocketAddr::new(bind_ip, 0);
            let socket = tokio::net::UdpSocket::bind(addr).await.map_err(|e| {
                crate::error::AgentError::Network(format!(
                    "failed to bind ephemeral port-mapping UDP listener on {bind_ip}: {e}"
                ))
            })?;
            let port = socket.local_addr().map(|a| a.port()).map_err(|e| {
                crate::error::AgentError::Network(format!(
                    "failed to read ephemeral port-mapping UDP local addr: {e}"
                ))
            })?;
            self.claim_published_port(deployment, owner_name, port)
                .await?;
            self.accumulate_port_map_udp_backend(port, owner_name, backend);
            let svc = Arc::new(UdpStreamService::new(
                Arc::clone(&self.port_map_registry),
                port,
                None,
            ));
            let handle = tokio::spawn(async move {
                if let Err(e) = svc.serve(socket).await {
                    tracing::error!(
                        port = port,
                        error = %e,
                        "port-mapping UDP stream proxy service failed"
                    );
                }
            });
            self.port_map_udp.write().await.insert(port, handle);
            info!(service = owner_name, port = port, bind = %bind_ip, backend = %backend,
                "Published port mapping on ephemeral host port (UDP)");
            Ok(port)
        }
    }

    /// Accumulate `backend` into the port-map UDP registry under `port`.
    fn accumulate_port_map_udp_backend(&self, port: u16, owner_name: &str, backend: SocketAddr) {
        if let Some(existing) = self.port_map_registry.resolve_udp(port) {
            let mut backends = existing.backends;
            if !backends.contains(&backend) {
                backends.push(backend);
            }
            self.port_map_registry.update_udp_backends(port, backends);
        } else {
            self.port_map_registry.register_udp(
                port,
                StreamService::new(owner_name.to_string(), vec![backend]),
            );
        }
    }

    /// Remove a port mapping's backend; free the listener + ownership when the
    /// last backend on that host port is gone. `port` is the bound host port
    /// (the caller passes the value returned by `publish_port_mapping`).
    pub async fn unpublish_port_mapping(
        &self,
        port: u16,
        protocol: PortProtocol,
        backend: SocketAddr,
    ) {
        match protocol {
            PortProtocol::Udp => self.unpublish_port_mapping_udp(port, backend).await,
            PortProtocol::Tcp => self.unpublish_port_mapping_tcp(port, backend).await,
        }
    }

    /// Drop `backend` from the port-map TCP service on `port`, freeing the
    /// listener when no backends remain.
    async fn unpublish_port_mapping_tcp(&self, port: u16, backend: SocketAddr) {
        let Some(existing) = self.port_map_registry.resolve_tcp(port) else {
            return;
        };
        let remaining: Vec<SocketAddr> = existing
            .backends
            .into_iter()
            .filter(|b| *b != backend)
            .collect();

        if remaining.is_empty() {
            let _ = self.port_map_registry.unregister_tcp(port);
            let mut listeners = self.port_map_tcp.write().await;
            if let Some(handle) = listeners.remove(&port) {
                handle.abort();
            }
            // Release host-port ownership so a different (deployment, service)
            // may bind it next.
            self.published_ports.write().await.remove(&port);
            // Close the host-firewall allow-rule now that the last backend is
            // gone (only here, so a sibling replica still using the port keeps
            // its rule). Best-effort.
            zlayer_overlay::firewall::remove_published_port(port, false);
            debug!(
                port = port,
                "Freed port-mapping TCP listener (no backends remain)"
            );
        } else {
            self.port_map_registry.update_tcp_backends(port, remaining);
        }
    }

    /// Drop `backend` from the port-map UDP service on `port`, freeing the
    /// listener when no backends remain.
    async fn unpublish_port_mapping_udp(&self, port: u16, backend: SocketAddr) {
        let Some(existing) = self.port_map_registry.resolve_udp(port) else {
            return;
        };
        let remaining: Vec<SocketAddr> = existing
            .backends
            .into_iter()
            .filter(|b| *b != backend)
            .collect();

        if remaining.is_empty() {
            let _ = self.port_map_registry.unregister_udp(port);
            let mut listeners = self.port_map_udp.write().await;
            if let Some(handle) = listeners.remove(&port) {
                handle.abort();
            }
            // Release host-port ownership so a different (deployment, service)
            // may bind it next.
            self.published_ports.write().await.remove(&port);
            // Close the host-firewall allow-rule now that the last backend is
            // gone (only here, so a sibling replica still using the port keeps
            // its rule). Best-effort.
            zlayer_overlay::firewall::remove_published_port(port, true);
            debug!(
                port = port,
                "Freed port-mapping UDP listener (no backends remain)"
            );
        } else {
            self.port_map_registry.update_udp_backends(port, remaining);
        }
    }

    /// Decide whether an endpoint should have a managed TLS certificate
    /// provisioned via ACME, returning the FQDN to provision when so.
    ///
    /// An endpoint qualifies ONLY when ALL of the following hold:
    /// 1. its protocol is [`Protocol::Https`],
    /// 2. it is publicly exposed ([`ExposeType::Public`]) — never for
    ///    internal-only endpoints (ACME HTTP-01 validation would fail and waste
    ///    Let's Encrypt rate limit), and
    /// 3. it carries a concrete, non-wildcard FQDN `host` (a real hostname,
    ///    not `None`, empty, `*`, or a `*.example.com` wildcard pattern — ACME
    ///    cannot satisfy an HTTP-01 challenge for a wildcard, and a hostless
    ///    endpoint has no name to validate).
    ///
    /// Returns the trimmed host string to provision, or `None` if the endpoint
    /// does not qualify. This is a pure function of the endpoint so it can be
    /// unit-tested without any ACME network access.
    #[must_use]
    pub fn endpoint_wants_managed_cert(endpoint: &zlayer_spec::EndpointSpec) -> Option<&str> {
        if endpoint.protocol != Protocol::Https {
            return None;
        }
        if endpoint.expose != ExposeType::Public {
            return None;
        }
        let host = endpoint.host.as_deref()?.trim();
        // Reject empty hosts and any wildcard pattern (`*`, `*.foo`, or a host
        // that contains a wildcard label). ACME HTTP-01 cannot validate these.
        if host.is_empty() || host.contains('*') {
            return None;
        }
        Some(host)
    }

    /// Add routes for a service based on its specification
    ///
    /// This creates proxy routes for each endpoint defined in the `ServiceSpec`.
    /// HTTP/HTTPS/WebSocket endpoints get L7 routes via the `ServiceRegistry`.
    /// TCP/UDP endpoints are tracked but their L4 registration is handled
    /// by the `ServiceManager::register_service_routes()` method.
    ///
    /// The owning `deployment` (from `ServiceSpec.deployment`) scopes the LB
    /// group keys so two deployments that share a `service`+`endpoint` name
    /// keep independent backend pools (Bug 7). `None` for standalone /
    /// single-deployment callers.
    #[allow(clippy::too_many_lines)]
    pub async fn add_service(&self, name: &str, spec: &ServiceSpec) {
        let deployment = spec.deployment.as_deref();
        let mut services = self.services.write().await;

        // Track which endpoints and ports we're adding
        let mut endpoint_names = Vec::new();
        let mut tcp_ports = Vec::new();
        let mut udp_ports = Vec::new();
        let mut http_ports = Vec::new();

        for endpoint in &spec.endpoints {
            match endpoint.protocol {
                Protocol::Http | Protocol::Https | Protocol::Websocket => {
                    // L7: register route in the ServiceRegistry
                    let entry = RouteEntry::from_endpoint(deployment, name, endpoint);
                    self.registry.register(entry).await;
                    http_ports.push(endpoint.port);

                    // Register one LB group per L7 endpoint, keyed by the
                    // deployment-scoped composite
                    // `{deployment}/{service}#{endpoint}`. This matches the
                    // `resolved.name` set by `RouteEntry::from_endpoint` and
                    // is required so that (a) different endpoints on the same
                    // service (potentially with different `target_role`
                    // filters) maintain independent backend pools, and (b)
                    // two deployments sharing a service+endpoint name do not
                    // cross-wire into one pool.
                    let lb_key = endpoint_lb_key(deployment, name, &endpoint.name);
                    self.load_balancer
                        .register(&lb_key, vec![], LbStrategy::RoundRobin);

                    info!(
                        service = name,
                        endpoint = %endpoint.name,
                        protocol = ?endpoint.protocol,
                        path = ?endpoint.path,
                        expose = ?endpoint.expose,
                        "Added HTTP proxy route for service"
                    );
                }
                Protocol::Tcp => {
                    tcp_ports.push(endpoint.port);
                    info!(
                        service = name,
                        endpoint = %endpoint.name,
                        protocol = ?endpoint.protocol,
                        port = endpoint.port,
                        expose = ?endpoint.expose,
                        "Tracking TCP stream endpoint for service"
                    );
                }
                Protocol::Udp => {
                    udp_ports.push(endpoint.port);
                    info!(
                        service = name,
                        endpoint = %endpoint.name,
                        protocol = ?endpoint.protocol,
                        port = endpoint.port,
                        expose = ?endpoint.expose,
                        "Tracking UDP stream endpoint for service"
                    );
                }
            }

            endpoint_names.push(endpoint.name.clone());
        }

        // Register a service-level LB group as well so legacy callers that
        // use `update_backends(service, ...)` (which fans out to all
        // endpoints) and any code that selects by bare service name still
        // resolve. Per-endpoint LB groups (registered above) are the
        // primary source for L7 select; this is a no-op for callers that
        // already use composite keys.
        self.load_balancer
            .register(name, vec![], LbStrategy::RoundRobin);

        services.insert(
            name.to_string(),
            ServiceTracking {
                deployment: deployment.map(str::to_string),
                endpoint_names,
                tcp_ports,
                udp_ports,
                http_ports,
            },
        );

        // Release the services write lock before touching ACME state below.
        drop(services);

        // ACME provisioning trigger: for every endpoint that wants a managed
        // certificate (public HTTPS vhost with a concrete FQDN), kick off a
        // detached, best-effort provisioning task. This is the keystone that
        // makes public HTTPS vhosts actually serve TLS — without it the ACME
        // engine exists but is never invoked, so the live `:443` listener has
        // no certificate for the host and serves "No certificate found".
        //
        // Each task is spawned (never awaited) so `add_service` does NOT block
        // on ACME, which can take seconds. `get_cert` is idempotent: it returns
        // a cached/on-disk cert without touching the network on a hit, and only
        // provisions via ACME HTTP-01 on a miss. The per-host
        // `provisioning_requested` guard ensures we fire at most once per host
        // for this manager's lifetime, so a reconcile re-running `add_service`
        // does not re-attempt a failed/in-flight provision and burn rate limit.
        if let Some(cert_manager) = &self.cert_manager {
            for endpoint in &spec.endpoints {
                let Some(host) = Self::endpoint_wants_managed_cert(endpoint) else {
                    continue;
                };
                let host = host.to_string();

                // De-dupe: only the first request for a given host schedules a task.
                {
                    let mut requested = self.provisioning_requested.write().await;
                    if !requested.insert(host.clone()) {
                        continue;
                    }
                }

                let cert_manager = Arc::clone(cert_manager);
                let sni_resolver = self.sni_resolver();
                let service_name = name.to_string();
                tokio::spawn(async move {
                    info!(
                        service = %service_name,
                        host = %host,
                        "Provisioning managed TLS certificate for public HTTPS vhost"
                    );
                    match cert_manager.get_cert(&host).await {
                        Ok((cert_pem, key_pem)) => {
                            // Hot-load into the SHARED resolver so the live
                            // listener serves it immediately.
                            if let Err(e) = sni_resolver.refresh_cert(&host, &cert_pem, &key_pem) {
                                warn!(
                                    host = %host,
                                    error = %e,
                                    "Provisioned certificate but failed to load it into the SNI resolver"
                                );
                            } else {
                                info!(
                                    host = %host,
                                    "Loaded managed TLS certificate into live HTTPS listener"
                                );
                            }
                        }
                        Err(e) => {
                            warn!(
                                service = %service_name,
                                host = %host,
                                error = %e,
                                "Failed to provision managed TLS certificate (best-effort; route is still registered)"
                            );
                        }
                    }
                });
            }
        }
    }

    /// Remove all routes, L4 listeners, and HTTP server handles for a service.
    ///
    /// This performs a full cleanup of all proxy resources associated with the
    /// service:
    /// - Removes L7 (HTTP/HTTPS/WebSocket) routes from the `ServiceRegistry`
    /// - Unregisters TCP/UDP stream services from the `StreamRegistry`
    /// - Removes port tracking for TCP/UDP listeners
    /// - Shuts down HTTP proxy server handles that were exclusively owned by
    ///   this service (only if no other service uses the same port)
    pub async fn remove_service(&self, name: &str) {
        let mut services = self.services.write().await;

        if let Some(tracking) = services.remove(name) {
            // 1. Remove L7 routes from the ServiceRegistry
            self.registry.unregister_service(name).await;

            // 1b. Remove from the load balancer (both the service-level
            //     group and every per-endpoint composite group). The
            //     per-endpoint keys are deployment-scoped to match the keys
            //     registered in `add_service`.
            self.load_balancer.unregister(name);
            let deployment = tracking.deployment.as_deref();
            for endpoint_name in &tracking.endpoint_names {
                let lb_key = endpoint_lb_key(deployment, name, endpoint_name);
                self.load_balancer.unregister(&lb_key);
            }

            // 2. Unregister TCP stream services and clear port tracking
            if !tracking.tcp_ports.is_empty() {
                let mut tcp_set = self.tcp_listeners.write().await;
                for port in &tracking.tcp_ports {
                    if let Some(registry) = &self.stream_registry {
                        let _ = registry.unregister_tcp(*port);
                    }
                    tcp_set.remove(port);
                    debug!(service = name, port = port, "Removed TCP listener tracking");
                }
            }

            // 3. Unregister UDP stream services and clear port tracking
            if !tracking.udp_ports.is_empty() {
                let mut udp_set = self.udp_listeners.write().await;
                for port in &tracking.udp_ports {
                    if let Some(registry) = &self.stream_registry {
                        let _ = registry.unregister_udp(*port);
                    }
                    udp_set.remove(port);
                    debug!(service = name, port = port, "Removed UDP listener tracking");
                }
            }

            // 4. Shut down HTTP proxy servers on ports exclusively owned by
            //    this service (skip ports still used by other services)
            if !tracking.http_ports.is_empty() {
                let ports_still_in_use: HashSet<u16> = services
                    .values()
                    .flat_map(|t| t.http_ports.iter().copied())
                    .collect();

                let mut servers = self.servers.write().await;
                for port in &tracking.http_ports {
                    if !ports_still_in_use.contains(port) {
                        if let Some(server) = servers.remove(port) {
                            server.shutdown();
                            info!(
                                service = name,
                                port = port,
                                "Shut down HTTP proxy server (no remaining services on port)"
                            );
                        }
                    }
                }
            }

            info!(service = name, "Removed all proxy resources for service");
        }
    }

    /// Add a single backend to a service.
    ///
    /// Adds to the service-level LB group **and** to every per-endpoint LB
    /// group tracked for `service`. Per-endpoint role filtering happens at
    /// collection time in the agent's service manager, so any backend
    /// surfaced here is already eligible for every endpoint.
    pub async fn add_backend(&self, service: &str, addr: SocketAddr) {
        self.registry.add_backend(service, addr).await;
        self.load_balancer.add_backend(service, addr);
        // Fan out to every per-endpoint LB group for backward-compat.
        let services = self.services.read().await;
        if let Some(tracking) = services.get(service) {
            let deployment = tracking.deployment.as_deref();
            for endpoint_name in &tracking.endpoint_names {
                let lb_key = endpoint_lb_key(deployment, service, endpoint_name);
                self.load_balancer.add_backend(&lb_key, addr);
            }
        }
        info!(service = service, backend = %addr, "Registered backend with proxy");
    }

    /// Remove a backend from a service.
    ///
    /// Removes from the service-level LB group **and** from every
    /// per-endpoint LB group.
    pub async fn remove_backend(&self, service: &str, addr: SocketAddr) {
        self.registry.remove_backend(service, addr).await;
        self.load_balancer.remove_backend(service, &addr);
        let services = self.services.read().await;
        if let Some(tracking) = services.get(service) {
            let deployment = tracking.deployment.as_deref();
            for endpoint_name in &tracking.endpoint_names {
                let lb_key = endpoint_lb_key(deployment, service, endpoint_name);
                self.load_balancer.remove_backend(&lb_key, &addr);
            }
        }
        debug!(service = service, backend = %addr, "Removed backend from service");
    }

    /// Update the health status of a backend in the load balancer.
    ///
    /// Delegates to [`LoadBalancer::mark_health`] so that unhealthy backends
    /// are skipped during selection. Health is tracked on both the
    /// service-level group and every per-endpoint group that contains
    /// this address.
    #[allow(clippy::unused_async)]
    pub async fn update_backend_health(&self, service: &str, addr: SocketAddr, healthy: bool) {
        self.load_balancer.mark_health(service, &addr, healthy);
        let services = self.services.read().await;
        if let Some(tracking) = services.get(service) {
            let deployment = tracking.deployment.as_deref();
            for endpoint_name in &tracking.endpoint_names {
                let lb_key = endpoint_lb_key(deployment, service, endpoint_name);
                self.load_balancer.mark_health(&lb_key, &addr, healthy);
            }
        }
        debug!(
            service = service,
            backend = %addr,
            healthy = healthy,
            "Updated backend health in load balancer"
        );
    }

    /// Update the backends for **every** endpoint of a service with the
    /// same list.
    ///
    /// Use this only when caller cannot distinguish per-endpoint backend
    /// sets (e.g., legacy paths that do not honor `target_role`). Prefer
    /// [`Self::update_endpoint_backends`] when per-endpoint filtering is
    /// possible.
    pub async fn update_backends(&self, service: &str, addrs: Vec<SocketAddr>) {
        self.registry.update_backends(service, addrs.clone()).await;
        // Update the service-level LB group plus every per-endpoint group.
        self.load_balancer.update_backends(service, addrs.clone());
        let services = self.services.read().await;
        if let Some(tracking) = services.get(service) {
            let deployment = tracking.deployment.as_deref();
            for endpoint_name in &tracking.endpoint_names {
                let lb_key = endpoint_lb_key(deployment, service, endpoint_name);
                self.load_balancer.update_backends(&lb_key, addrs.clone());
            }
        }
        debug!(service = service, "Updated backends for service");
    }

    /// Update backends for a single L7 endpoint of a service.
    ///
    /// This honors [`EndpointSpec::target_role`] filtering: the caller
    /// supplies the role-filtered backend list and this method updates
    /// only the routes and LB group corresponding to `(service,
    /// endpoint_name)`.
    pub async fn update_endpoint_backends(
        &self,
        service: &str,
        endpoint_name: &str,
        addrs: Vec<SocketAddr>,
    ) {
        self.registry
            .update_backends_for_endpoint(service, endpoint_name, addrs.clone())
            .await;
        // Resolve the owning deployment so the LB key matches what
        // `add_service` registered.
        let deployment = {
            let services = self.services.read().await;
            services.get(service).and_then(|t| t.deployment.clone())
        };
        let lb_key = endpoint_lb_key(deployment.as_deref(), service, endpoint_name);
        self.load_balancer.update_backends(&lb_key, addrs);
        debug!(
            service = service,
            endpoint = endpoint_name,
            "Updated backends for service endpoint"
        );
    }

    /// Get the number of registered routes
    pub async fn route_count(&self) -> usize {
        self.registry.route_count().await
    }

    /// Get the list of registered service names
    pub async fn list_services(&self) -> Vec<String> {
        self.services.read().await.keys().cloned().collect()
    }

    /// Check if a service has any registered endpoints
    pub async fn has_service(&self, name: &str) -> bool {
        self.services.read().await.contains_key(name)
    }
}

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

    fn mock_service_spec_with_endpoints() -> ServiceSpec {
        use zlayer_spec::*;
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: http
        protocol: http
        port: 8080
        path: /api
        expose: public
      - name: websocket
        protocol: websocket
        port: 8081
        path: /ws
        expose: internal
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap()
    }

    fn mock_service_spec_tcp_only() -> ServiceSpec {
        mock_service_spec_tcp_only_port(9000)
    }

    fn mock_service_spec_tcp_only_port(port: u16) -> ServiceSpec {
        use zlayer_spec::*;
        let yaml = format!(
            "
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: grpc
        protocol: tcp
        port: {port}
"
        );
        serde_yaml::from_str::<DeploymentSpec>(&yaml)
            .unwrap()
            .services
            .remove("test")
            .unwrap()
    }

    /// Reserve an unused localhost TCP port by binding a listener on `:0`,
    /// reading the assigned port, and dropping the listener.
    ///
    /// There is an inherent race between dropping the listener and the test
    /// re-binding the port, but this is dramatically more reliable than
    /// hard-coding a port (e.g., 9000) which is commonly in use on dev
    /// machines (php-fpm, the running zlayer daemon, etc.).
    fn reserve_free_tcp_port() -> u16 {
        let listener =
            std::net::TcpListener::bind("127.0.0.1:0").expect("failed to bind ephemeral test port");
        listener.local_addr().unwrap().port()
    }

    #[tokio::test]
    async fn test_proxy_manager_new() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        assert_eq!(manager.route_count().await, 0);
        assert!(manager.list_services().await.is_empty());
    }

    #[tokio::test]
    async fn test_add_service_with_http_endpoints() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_with_endpoints();
        manager.add_service("api", &spec).await;

        // Should have 2 routes (http and websocket)
        assert_eq!(manager.route_count().await, 2);
        assert!(manager.has_service("api").await);
    }

    #[tokio::test]
    async fn test_tcp_endpoints_tracked_not_routed() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_tcp_only();
        manager.add_service("grpc-service", &spec).await;

        // TCP endpoints don't add HTTP routes
        assert_eq!(manager.route_count().await, 0);
        // But the service is still tracked with its endpoint name
        assert!(manager.has_service("grpc-service").await);
    }

    #[tokio::test]
    async fn test_remove_service() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_with_endpoints();
        manager.add_service("api", &spec).await;
        assert_eq!(manager.route_count().await, 2);

        manager.remove_service("api").await;
        assert_eq!(manager.route_count().await, 0);
        assert!(!manager.has_service("api").await);
    }

    #[tokio::test]
    async fn test_backend_management() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry.clone(), None);

        let spec = mock_service_spec_with_endpoints();
        manager.add_service("api", &spec).await;

        // Add backends
        let addr1: SocketAddr = "127.0.0.1:8080".parse().unwrap();
        let addr2: SocketAddr = "127.0.0.1:8081".parse().unwrap();

        manager.add_backend("api", addr1).await;
        manager.add_backend("api", addr2).await;

        // Verify backends via the registry's resolve
        let resolved = registry.resolve(None, "/api").await.unwrap();
        assert_eq!(resolved.backends.len(), 2);

        // Remove a backend
        manager.remove_backend("api", addr1).await;
        let resolved = registry.resolve(None, "/api").await.unwrap();
        assert_eq!(resolved.backends.len(), 1);
    }

    #[tokio::test]
    async fn test_update_backends_replaces_all() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry.clone(), None);

        let spec = mock_service_spec_with_endpoints();
        manager.add_service("api", &spec).await;

        // Add initial backend
        let addr1: SocketAddr = "127.0.0.1:8080".parse().unwrap();
        manager.add_backend("api", addr1).await;

        // Update with new backends (replaces)
        let new_backends: Vec<SocketAddr> = vec![
            "127.0.0.1:9000".parse().unwrap(),
            "127.0.0.1:9001".parse().unwrap(),
            "127.0.0.1:9002".parse().unwrap(),
        ];
        manager.update_backends("api", new_backends).await;

        let resolved = registry.resolve(None, "/api").await.unwrap();
        assert_eq!(resolved.backends.len(), 3);
    }

    #[tokio::test]
    async fn test_config_builder() {
        let config = ProxyManagerConfig::new("0.0.0.0:8080".parse().unwrap())
            .with_https("0.0.0.0:8443".parse().unwrap())
            .with_http2(false);

        assert_eq!(
            config.http_addr,
            "0.0.0.0:8080".parse::<SocketAddr>().unwrap()
        );
        assert_eq!(
            config.https_addr,
            Some("0.0.0.0:8443".parse::<SocketAddr>().unwrap())
        );
        assert!(!config.http2_enabled);
    }

    /// Test that `ensure_ports_for_service` correctly differentiates
    /// Public (0.0.0.0) vs Internal (overlay or 127.0.0.1) bind addresses.
    /// We can't actually bind in unit tests, but we verify the function
    /// processes both endpoint types without error.
    #[tokio::test]
    async fn test_ensure_ports_differentiates_public_and_internal() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_with_endpoints();
        // Passing None for overlay_ip: internal endpoints should fall back to 127.0.0.1
        let result = manager.ensure_ports_for_service(&spec, None).await;
        // listen_on may fail because we can't actually bind in tests, but
        // the function itself should run without panicking.
        let _ = result;
    }

    #[tokio::test]
    async fn test_ensure_ports_with_overlay_ip() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_with_endpoints();
        // Pass an overlay IP -- internal endpoints should bind there
        let overlay_ip: IpAddr = "10.200.0.5".parse().unwrap();
        let result = manager
            .ensure_ports_for_service(&spec, Some(overlay_ip))
            .await;
        let _ = result;
    }

    fn mock_mixed_service_spec() -> ServiceSpec {
        use zlayer_spec::*;
        serde_yaml::from_str::<DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  mixed:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: http
        protocol: http
        port: 8080
        path: /api
        expose: public
      - name: grpc
        protocol: tcp
        port: 9000
        expose: public
      - name: game
        protocol: udp
        port: 27015
        expose: public
",
        )
        .unwrap()
        .services
        .remove("mixed")
        .unwrap()
    }

    #[tokio::test]
    async fn test_add_mixed_service_tracks_all_endpoints() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_mixed_service_spec();
        manager.add_service("mixed", &spec).await;

        // Only 1 HTTP route (tcp and udp don't add HTTP routes)
        assert_eq!(manager.route_count().await, 1);
        // Service is tracked
        assert!(manager.has_service("mixed").await);
    }

    #[tokio::test]
    async fn test_ensure_ports_tcp_with_stream_registry() {
        use zlayer_proxy::StreamService;

        // `reserve_free_tcp_port` binds `127.0.0.1:0`, reads the assigned port,
        // then drops the listener — so there is a TOCTOU window where, under
        // parallel test threads, another process/test can grab that port before
        // `ensure_ports_for_service` rebinds it. When that happens the bind is
        // skipped and the port is never tracked. Retry with a fresh OS-assigned
        // port (collisions are rare) so this test is deterministic.
        let mut bound = false;
        for _ in 0..16 {
            let stream_registry = Arc::new(StreamRegistry::new());
            let config = ProxyManagerConfig::default();
            let registry = Arc::new(ServiceRegistry::new());
            let mut manager = ProxyManager::new(config, registry, None);
            manager.set_stream_registry(stream_registry.clone());

            let port = reserve_free_tcp_port();
            let spec = mock_service_spec_tcp_only_port(port);

            // Register the TCP service in the stream registry first (as ServiceManager does)
            stream_registry
                .register_tcp(port, StreamService::new("grpc-service".to_string(), vec![]));

            // Ensure ports -- should bind TCP listener
            let result = manager.ensure_ports_for_service(&spec, None).await;
            assert!(result.is_ok());

            // Verify the TCP listener port is tracked; retry on a lost-the-port race.
            if manager.tcp_listeners.read().await.contains(&port) {
                bound = true;
                break;
            }
        }
        assert!(
            bound,
            "ensure_ports_for_service never tracked an OS-assigned TCP port across 16 attempts"
        );
    }

    #[tokio::test]
    async fn test_ensure_ports_tcp_without_stream_registry() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_service_spec_tcp_only();

        // Without stream registry, ensure_ports should not fail, just warn
        let result = manager.ensure_ports_for_service(&spec, None).await;
        assert!(result.is_ok());

        // No TCP listeners should be tracked
        let tcp_ports = manager.tcp_listeners.read().await;
        assert!(tcp_ports.is_empty());
    }

    #[tokio::test]
    async fn test_stream_registry_setter() {
        let stream_registry = Arc::new(StreamRegistry::new());
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let mut manager = ProxyManager::new(config, registry, None);

        assert!(manager.stream_registry().is_none());
        manager.set_stream_registry(stream_registry.clone());
        assert!(manager.stream_registry().is_some());
    }

    /// Single-member service spec with one INTERNAL TCP endpoint published on
    /// `port`. Internal (not Public) so the loopback path actually binds it.
    fn mock_internal_tcp_spec(port: u16) -> ServiceSpec {
        use zlayer_spec::*;
        let yaml = format!(
            "
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    scale:
      mode: fixed
      replicas: 1
    endpoints:
      - name: tcp
        protocol: tcp
        port: {port}
        expose: internal
"
        );
        serde_yaml::from_str::<DeploymentSpec>(&yaml)
            .unwrap()
            .services
            .remove("test")
            .unwrap()
    }

    /// End-to-end loopback publish: spin up a real backend `TcpListener`,
    /// publish it on the node loopback, connect to `127.0.0.1:<publish_port>`
    /// and assert bytes round-trip through the forward; then unpublish and
    /// assert the port is freed (a fresh bind succeeds).
    #[tokio::test]
    async fn test_publish_loopback_round_trips_then_frees_port() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        // Real backend that echoes a single line back with a known reply.
        let backend = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let backend_addr = backend.local_addr().unwrap();
        let backend_ip = backend_addr.ip();
        let backend_port = backend_addr.port();
        tokio::spawn(async move {
            if let Ok((mut sock, _)) = backend.accept().await {
                let mut buf = [0u8; 16];
                let n = sock.read(&mut buf).await.unwrap_or(0);
                // Echo back what we received, prefixed.
                let _ = sock.write_all(b"pong:").await;
                let _ = sock.write_all(&buf[..n]).await;
                let _ = sock.flush().await;
            }
        });

        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Reserve a free publish port (the node-loopback address).
        let publish_port = reserve_free_tcp_port();
        let spec = mock_internal_tcp_spec(publish_port);
        assert!(
            spec.publish_to_node_loopback(),
            "single-member internal spec should publish to loopback"
        );

        // The backend is the real listener; port_override forces the forward
        // target to the backend's actual ephemeral port (the macOS-style path).
        manager
            .publish_loopback_for_container(
                Some("dep-a"),
                "test",
                &spec,
                backend_ip,
                Some(backend_port),
            )
            .await
            .expect("publish should succeed on a free port");

        // Connect to 127.0.0.1:<publish_port> and round-trip a payload.
        let mut client = tokio::net::TcpStream::connect((Ipv4Addr::LOCALHOST, publish_port))
            .await
            .expect("connect to published loopback port");
        client.write_all(b"ping").await.unwrap();
        client.flush().await.unwrap();
        let mut reply = Vec::new();
        client.read_to_end(&mut reply).await.unwrap();
        assert_eq!(&reply, b"pong:ping");
        drop(client);

        // Unpublish; the last backend's removal frees the listener.
        manager
            .unpublish_loopback_for_container(&spec, backend_ip, Some(backend_port))
            .await;

        // The aborted accept task drops the listener asynchronously; retry a
        // few times so the OS reclaims the port before we assert it is free.
        let mut bound = None;
        for _ in 0..50 {
            match std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, publish_port)) {
                Ok(l) => {
                    bound = Some(l);
                    break;
                }
                Err(_) => tokio::time::sleep(Duration::from_millis(20)).await,
            }
        }
        assert!(
            bound.is_some(),
            "loopback port {publish_port} should be freed after unpublish"
        );
    }

    #[tokio::test]
    async fn test_publish_loopback_skips_public_endpoints() {
        // Public endpoints are already on 0.0.0.0, so the loopback path must
        // NOT bind 127.0.0.1:<port> again. mock_mixed_service_spec exposes
        // everything as public.
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = mock_mixed_service_spec();
        let backend_ip: IpAddr = "127.0.0.1".parse().unwrap();
        manager
            .publish_loopback_for_container(Some("dep-a"), "mixed", &spec, backend_ip, None)
            .await
            .expect("public-only spec publishes nothing and must not error");

        // No loopback listeners should have been created for public endpoints.
        assert!(manager.loopback_tcp.read().await.is_empty());
        assert!(manager.loopback_udp.read().await.is_empty());
    }

    #[tokio::test]
    async fn test_registry_accessor() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry.clone(), None);

        // registry() should return the same Arc
        assert_eq!(Arc::as_ptr(&manager.registry()), Arc::as_ptr(&registry));
    }

    /// Bug 7: a host port published by deployment A must NOT be cross-wired
    /// into deployment B's backend pool. B's publish on the same port is
    /// REFUSED with `PortConflict`, and `:<port>` keeps resolving to A's
    /// backend only.
    #[tokio::test]
    async fn test_published_port_ownership_rejects_cross_deployment() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Reserve a free publish port shared by both deployments.
        let publish_port = reserve_free_tcp_port();
        let spec = mock_internal_tcp_spec(publish_port);

        // Distinct container backends for the two deployments.
        let backend_a: IpAddr = "10.0.0.1".parse().unwrap();
        let tgt_a = 5001u16;
        let backend_b: IpAddr = "10.0.0.2".parse().unwrap();
        let tgt_b = 5002u16;

        // Deployment A claims the port -> succeeds.
        manager
            .publish_loopback_for_container(Some("dep-a"), "svc", &spec, backend_a, Some(tgt_a))
            .await
            .expect("deployment A should claim the free port");

        // Deployment B publishing the SAME port -> REFUSED.
        let err = manager
            .publish_loopback_for_container(Some("dep-b"), "svc", &spec, backend_b, Some(tgt_b))
            .await
            .expect_err("deployment B must be refused on an owned port");
        match err {
            crate::error::AgentError::PortConflict { port, .. } => {
                assert_eq!(port, publish_port);
            }
            other => panic!("expected PortConflict, got {other:?}"),
        }

        // `:<port>` must still serve ONLY deployment A's backend — B was never
        // appended into the foreign pool.
        let svc = manager
            .loopback_registry
            .resolve_tcp(publish_port)
            .expect("port should still be registered to deployment A");
        let expected_a = SocketAddr::new(backend_a, tgt_a);
        let foreign_b = SocketAddr::new(backend_b, tgt_b);
        assert_eq!(svc.backends, vec![expected_a]);
        assert!(
            !svc.backends.contains(&foreign_b),
            "deployment B's backend must NOT be cross-wired into the pool"
        );
    }

    /// A second replica of the SAME (deployment, service) on an already-owned
    /// port is a legitimate scale-up: the replica backend IS appended.
    #[tokio::test]
    async fn test_published_port_same_owner_appends_replica() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let publish_port = reserve_free_tcp_port();
        let spec = mock_internal_tcp_spec(publish_port);

        let replica1: IpAddr = "10.0.0.1".parse().unwrap();
        let replica2: IpAddr = "10.0.0.2".parse().unwrap();
        let target_port = 6000u16;

        // First replica claims the port.
        manager
            .publish_loopback_for_container(
                Some("dep-a"),
                "svc",
                &spec,
                replica1,
                Some(target_port),
            )
            .await
            .expect("first replica claims the port");

        // Second replica of the SAME (deployment, service) -> appended.
        manager
            .publish_loopback_for_container(
                Some("dep-a"),
                "svc",
                &spec,
                replica2,
                Some(target_port),
            )
            .await
            .expect("same-owner second replica should be accepted");

        let svc = manager
            .loopback_registry
            .resolve_tcp(publish_port)
            .expect("port should be registered");
        let b1 = SocketAddr::new(replica1, target_port);
        let b2 = SocketAddr::new(replica2, target_port);
        assert_eq!(svc.backends.len(), 2, "both replicas should be in the pool");
        assert!(svc.backends.contains(&b1));
        assert!(svc.backends.contains(&b2));
    }

    /// After the owning service unpublishes its last backend, the host-port
    /// ownership entry is released so a different (deployment, service) may
    /// claim it.
    #[tokio::test]
    async fn test_published_port_freed_on_unpublish() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let publish_port = reserve_free_tcp_port();
        let spec = mock_internal_tcp_spec(publish_port);
        let backend_a: IpAddr = "10.0.0.1".parse().unwrap();
        let target_port = 7000u16;

        manager
            .publish_loopback_for_container(
                Some("dep-a"),
                "svc",
                &spec,
                backend_a,
                Some(target_port),
            )
            .await
            .expect("deployment A claims the port");
        assert!(manager
            .published_ports
            .read()
            .await
            .contains_key(&publish_port));

        // Unpublish A's only backend -> ownership released.
        manager
            .unpublish_loopback_for_container(&spec, backend_a, Some(target_port))
            .await;
        assert!(
            !manager
                .published_ports
                .read()
                .await
                .contains_key(&publish_port),
            "ownership entry should be cleared once the last backend is gone"
        );

        // A different deployment can now claim the freed port.
        let backend_b: IpAddr = "10.0.0.2".parse().unwrap();
        manager
            .publish_loopback_for_container(
                Some("dep-b"),
                "svc",
                &spec,
                backend_b,
                Some(target_port),
            )
            .await
            .expect("freed port should be claimable by another deployment");
    }

    #[tokio::test]
    #[allow(clippy::similar_names)]
    async fn test_start_ingress_is_idempotent() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Use free ephemeral ports so the test does not need root to bind
        // 80/443. No CertManager is configured, so only the HTTP listener
        // registers (HTTPS warns + returns early).
        let http_port = reserve_free_tcp_port();
        let https_port = reserve_free_tcp_port();

        manager.start_ingress_on(http_port, https_port).await;
        // The HTTP ingress server should be registered on its port.
        assert!(
            manager.servers.read().await.contains_key(&http_port),
            "HTTP ingress should be registered"
        );
        assert!(
            manager.ingress_started.load(Ordering::SeqCst),
            "ingress_started flag should be set"
        );
        let count_after_first = manager.servers.read().await.len();

        // Second call is a no-op: the idempotency guard short-circuits, so the
        // server map does not grow.
        manager.start_ingress_on(http_port, https_port).await;
        assert_eq!(
            manager.servers.read().await.len(),
            count_after_first,
            "second start_ingress call must not register additional servers"
        );
    }

    /// Regression: the `:80` HTTP ingress server MUST carry the manager's
    /// `CertManager` so it can serve ACME HTTP-01 challenges
    /// (`/.well-known/acme-challenge/<token>`), which always arrive on `:80`.
    ///
    /// Without this wiring the challenge interception is skipped (its guard is
    /// `if let Some(cm) = self.cert_manager`) and the request falls through to
    /// the vhost match, which 403s — so certs can never be issued. The
    /// server-level `test_acme_challenge_served_not_denied` did NOT catch this
    /// because it builds the `ProxyServer` with a cert manager directly; the
    /// gap was in `proxy_manager` failing to wire it into the `:80` server.
    ///
    /// Network-free: builds the server via the extracted helper without binding
    /// any port.
    #[tokio::test]
    async fn test_http_ingress_carries_cert_manager_for_acme() {
        let tmp = tempfile::tempdir().unwrap();
        let cm = Arc::new(
            CertManager::new(tmp.path().to_string_lossy().into_owned(), None)
                .await
                .unwrap(),
        );

        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, Some(Arc::clone(&cm)));

        let http_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 80);
        let http_server = manager.build_http_ingress_server(http_addr, None);

        assert!(
            http_server.cert_manager().is_some(),
            "the :80 HTTP ingress server must carry the CertManager so ACME \
             HTTP-01 challenges (which always arrive on :80) are served, not 403'd"
        );
    }

    /// The HTTP ingress server is still built when NO `CertManager` is
    /// configured — only the `with_cert_manager` call is gated on `Some`, never
    /// the whole `:80` ingress. (A non-ACME deployment still needs `:80`.)
    #[tokio::test]
    async fn test_http_ingress_built_without_cert_manager() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let http_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 80);
        let http_server = manager.build_http_ingress_server(http_addr, None);

        assert!(
            http_server.cert_manager().is_none(),
            "without a configured CertManager the :80 server carries none"
        );
    }

    #[tokio::test]
    async fn publish_port_mapping_binds_explicit_host_port() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Grab a free explicit port deterministically: bind, read, drop.
        let port = {
            let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            l.local_addr().unwrap().port()
        };

        let mapping = PortMapping {
            host_port: Some(port),
            container_port: 8080,
            protocol: PortProtocol::Tcp,
            host_ip: "127.0.0.1".to_string(),
        };
        // Dummy backend; the connect below only exercises the accept side.
        let backend: SocketAddr = "127.0.0.1:1".parse().unwrap();

        let bound = manager
            .publish_port_mapping(Some("dep"), "svc", &mapping, backend)
            .await
            .expect("publish should succeed");
        assert_eq!(bound, port, "explicit host port must be honored");

        // A real host listener now exists: a TCP connect must be accepted.
        let conn = tokio::net::TcpStream::connect(("127.0.0.1", bound)).await;
        assert!(conn.is_ok(), "expected a live host listener on {bound}");
    }

    #[tokio::test]
    async fn publish_port_mapping_rejects_cross_owner_explicit_port() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Free explicit port: bind, read, drop.
        let port = {
            let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
            l.local_addr().unwrap().port()
        };

        let mapping = PortMapping {
            host_port: Some(port),
            container_port: 8080,
            protocol: PortProtocol::Tcp,
            host_ip: "127.0.0.1".to_string(),
        };
        let backend_a: SocketAddr = "127.0.0.1:1".parse().unwrap();
        let backend_b: SocketAddr = "127.0.0.1:2".parse().unwrap();

        // Owner A claims the port.
        manager
            .publish_port_mapping(Some("depA"), "ownerA", &mapping, backend_a)
            .await
            .expect("owner A publish should succeed");

        // Owner B on the SAME explicit port must be refused.
        let err = manager
            .publish_port_mapping(Some("depB"), "ownerB", &mapping, backend_b)
            .await
            .expect_err("cross-owner explicit port must be refused");
        assert!(
            matches!(err, crate::error::AgentError::PortConflict { .. }),
            "expected PortConflict, got {err:?}"
        );
    }

    #[tokio::test]
    async fn publish_port_mapping_ephemeral_returns_nonzero() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let mapping = PortMapping {
            host_port: None,
            container_port: 8080,
            protocol: PortProtocol::Tcp,
            host_ip: "127.0.0.1".to_string(),
        };
        let backend: SocketAddr = "127.0.0.1:1".parse().unwrap();

        let bound = manager
            .publish_port_mapping(Some("dep"), "svc", &mapping, backend)
            .await
            .expect("ephemeral publish should succeed");
        assert_ne!(bound, 0, "ephemeral publish must resolve to a real port");
    }

    // --- Activator + RPS provider ------------------------------------------

    /// A `ScaleTrigger` that records its calls and registers a healthy backend
    /// on the shared load balancer, simulating the `ServiceManager` bringing a
    /// scaled-to-zero service up.
    struct FakeScaleTrigger {
        lb: Arc<LoadBalancer>,
        lb_key: String,
        calls: Arc<std::sync::Mutex<Vec<(String, u32)>>>,
    }

    #[async_trait::async_trait]
    impl ScaleTrigger for FakeScaleTrigger {
        async fn scale_to(&self, service: &str, replicas: u32) -> std::result::Result<(), String> {
            self.calls
                .lock()
                .unwrap()
                .push((service.to_string(), replicas));
            // Bring the LB group up with a healthy backend.
            self.lb.register(
                &self.lb_key,
                vec!["127.0.0.1:9".parse().unwrap()],
                LbStrategy::RoundRobin,
            );
            Ok(())
        }
    }

    #[test]
    fn service_name_from_key_strips_deployment_and_endpoint() {
        assert_eq!(
            ServiceActivator::service_name_from_key("dep/api#http"),
            "api"
        );
        assert_eq!(ServiceActivator::service_name_from_key("api#http"), "api");
        assert_eq!(ServiceActivator::service_name_from_key("dep/api"), "api");
        assert_eq!(ServiceActivator::service_name_from_key("api"), "api");
    }

    #[tokio::test]
    async fn service_activator_scales_and_waits_for_backend() {
        let lb = Arc::new(LoadBalancer::new());
        let lb_key = "dep/api#http".to_string();
        // Idle group: present but no backend.
        lb.register(&lb_key, vec![], LbStrategy::RoundRobin);

        let calls = Arc::new(std::sync::Mutex::new(Vec::new()));
        let trigger = Arc::new(FakeScaleTrigger {
            lb: Arc::clone(&lb),
            lb_key: lb_key.clone(),
            calls: Arc::clone(&calls),
        });
        let activator = ServiceActivator::new(trigger, Arc::clone(&lb));

        activator
            .activate(&lb_key)
            .await
            .expect("activation should succeed once a backend is registered");

        let recorded = calls.lock().unwrap().clone();
        assert_eq!(
            recorded,
            vec![("api".to_string(), DEFAULT_ACTIVATION_FLOOR)],
            "scale_to should be called with the bare service name and the floor"
        );
    }

    #[tokio::test]
    async fn rps_registry_provider_reflects_recorded_requests() {
        let reg = Arc::new(RpsRegistry::new());
        let provider = RpsRegistryProvider::new(Arc::clone(&reg));

        assert!((provider.rps("svc").await - 0.0).abs() < f64::EPSILON);

        reg.record("svc").await;
        reg.record("svc").await;
        assert!(
            provider.rps("svc").await > 0.0,
            "provider must reflect recorded requests"
        );
    }

    #[tokio::test]
    async fn rps_registry_accessor_is_shared() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let reg = manager.rps_registry();
        reg.record("svc").await;
        // The provider built from the manager observes the same underlying data.
        let provider = manager.rps_provider();
        assert!(provider.rps("svc").await > 0.0);
    }

    #[tokio::test]
    async fn set_activator_is_installed() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        assert!(manager.current_activator().await.is_none());

        let lb = manager.load_balancer();
        let trigger = Arc::new(FakeScaleTrigger {
            lb: Arc::clone(&lb),
            lb_key: "dep/api#http".to_string(),
            calls: Arc::new(std::sync::Mutex::new(Vec::new())),
        });
        manager.install_service_activator(trigger).await;

        assert!(
            manager.current_activator().await.is_some(),
            "activator should be installed after install_service_activator"
        );
    }

    /// Build an `EndpointSpec` with the given protocol/expose/host for the
    /// managed-cert gating tests. All other fields are defaults.
    fn cert_test_endpoint(
        protocol: Protocol,
        expose: ExposeType,
        host: Option<&str>,
    ) -> zlayer_spec::EndpointSpec {
        zlayer_spec::EndpointSpec {
            name: "web".to_string(),
            protocol,
            port: 443,
            target_port: None,
            path: None,
            host: host.map(str::to_string),
            expose,
            stream: None,
            target_role: None,
            tunnel: None,
        }
    }

    #[test]
    fn endpoint_wants_managed_cert_gates_correctly() {
        // Qualifies: public HTTPS with a concrete FQDN.
        assert_eq!(
            ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                Protocol::Https,
                ExposeType::Public,
                Some("console.zatabase.io"),
            )),
            Some("console.zatabase.io")
        );

        // Host is trimmed.
        assert_eq!(
            ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                Protocol::Https,
                ExposeType::Public,
                Some("  console.zatabase.io  "),
            )),
            Some("console.zatabase.io")
        );

        // Rejected: internal exposure (ACME HTTP-01 would fail).
        assert!(
            ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                Protocol::Https,
                ExposeType::Internal,
                Some("console.zatabase.io"),
            ))
            .is_none()
        );

        // Rejected: not HTTPS.
        for proto in [
            Protocol::Http,
            Protocol::Websocket,
            Protocol::Tcp,
            Protocol::Udp,
        ] {
            assert!(
                ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                    proto,
                    ExposeType::Public,
                    Some("console.zatabase.io"),
                ))
                .is_none()
            );
        }

        // Rejected: no host.
        assert!(
            ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                Protocol::Https,
                ExposeType::Public,
                None,
            ))
            .is_none()
        );

        // Rejected: empty / whitespace host.
        assert!(
            ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                Protocol::Https,
                ExposeType::Public,
                Some("   "),
            ))
            .is_none()
        );

        // Rejected: wildcard patterns (ACME HTTP-01 cannot satisfy these).
        for wildcard in ["*", "*.zatabase.io", "console.*.io"] {
            assert!(
                ProxyManager::endpoint_wants_managed_cert(&cert_test_endpoint(
                    Protocol::Https,
                    ExposeType::Public,
                    Some(wildcard),
                ))
                .is_none()
            );
        }
    }

    #[tokio::test]
    async fn sni_resolver_accessor_is_shared_and_stable() {
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        // Every call returns a clone of the SAME Arc the live listeners use, so
        // a cert hot-loaded into one handle is visible through any other.
        let a = manager.sni_resolver();
        let b = manager.sni_resolver();
        assert!(
            Arc::ptr_eq(&a, &b),
            "sni_resolver() must return clones of one shared resolver"
        );
        assert_eq!(a.cert_count(), 0);
    }

    #[tokio::test]
    async fn add_service_without_cert_manager_does_not_panic_or_provision() {
        // No CertManager configured: the provisioning trigger must be a no-op
        // (and must never reach the network), even for a public HTTPS vhost.
        let config = ProxyManagerConfig::default();
        let registry = Arc::new(ServiceRegistry::new());
        let manager = ProxyManager::new(config, registry, None);

        let spec = serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  web:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: web
        protocol: https
        port: 443
        host: console.zatabase.io
        expose: public
",
        )
        .unwrap()
        .services
        .remove("web")
        .unwrap();

        manager.add_service("web", &spec).await;
        // Route is still registered even though no cert manager exists.
        assert!(manager.has_service("web").await);
        // Shared resolver is untouched (no provisioning attempted).
        assert_eq!(manager.sni_resolver().cert_count(), 0);
    }
}