eggress-cli 1.0.4

use std::net::SocketAddr;
use std::process::ExitCode;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Duration;

#[cfg(feature = "pproxy-compat")]
use eggress_cli::EXIT_UNSUPPORTED_FEATURE;
use eggress_cli::{
    EXIT_CLI_PARSE_ERROR, EXIT_CONFIG_VALIDATION, EXIT_RUNTIME_FAILURE, EXIT_SIGINT, EXIT_SIGTERM,
    EXIT_SUCCESS,
};

use clap::{Parser, Subcommand};
use eggress_core::listener::{TcpListener, TcpListenerConfig};
use eggress_routing::{RouteActionSpec, RouteService, Router, SharedRoutingService};
use eggress_server::ConnectionConfig;
use tokio_util::sync::CancellationToken;
use tracing_subscriber::{fmt, EnvFilter};

static CONNECTION_COUNTER: AtomicU64 = AtomicU64::new(1);
static ACTIVE_CONNECTIONS: AtomicU64 = AtomicU64::new(0);
/// Notified when an in-flight connection finishes so the shutdown drain loop
/// can react without polling the counter on a 100 ms timer.
static ACTIVE_CONNECTIONS_DRAIN: std::sync::LazyLock<tokio::sync::Notify> =
    std::sync::LazyLock::new(tokio::sync::Notify::new);

#[derive(Parser, Debug)]
#[command(name = "eggress", version, about = "A multi-protocol TCP proxy")]
struct Cli {
    #[arg(short = 'l', long = "listen", value_name = "URI")]
    listeners: Vec<String>,

    #[arg(short = 'r', long = "remote", value_name = "URI")]
    upstreams: Vec<String>,

    #[arg(long = "log-format", value_name = "FORMAT", default_value = "pretty")]
    log_format: String,

    #[arg(long = "config", value_name = "PATH")]
    config: Option<String>,

    #[arg(long = "rules-file", value_name = "PATH")]
    rules_file: Option<String>,

    #[command(subcommand)]
    command: Option<SubCommand>,
}

#[derive(Subcommand, Debug)]
enum SubCommand {
    Route(RouteExplain),
    Upstream(UpstreamCommand),
    #[cfg(feature = "pproxy-compat")]
    Pproxy(PproxyCommand),
    #[cfg(feature = "operations")]
    SystemProxy(SystemProxyCommand),
}

#[cfg(feature = "operations")]
#[derive(Parser, Debug)]
struct SystemProxyCommand {
    #[command(subcommand)]
    action: SystemProxyAction,
}

#[cfg(feature = "operations")]
#[derive(Subcommand, Debug)]
enum SystemProxyAction {
    /// Inspect current system proxy settings (read-only)
    Inspect(SystemProxyInspect),
}

#[cfg(feature = "operations")]
#[derive(Parser, Debug)]
struct SystemProxyInspect {
    /// Output as JSON
    #[arg(long)]
    json: bool,
}

#[derive(Parser, Debug)]
struct UpstreamCommand {
    #[command(subcommand)]
    action: UpstreamAction,
}

#[derive(Subcommand, Debug)]
enum UpstreamAction {
    Test(UpstreamTest),
}

#[derive(Parser, Debug)]
struct UpstreamTest {
    #[arg(short, long, value_name = "ID")]
    id: Option<String>,

    #[arg(short, long, value_name = "HOST:PORT")]
    target: Option<String>,

    #[arg(short, long)]
    config: Option<String>,

    #[arg(long, default_value = "5")]
    timeout: u64,

    #[arg(long, default_value = "proxy")]
    mode: String,

    #[arg(long)]
    json: bool,
}

#[derive(Parser, Debug)]
struct RouteExplain {
    target: String,

    #[arg(short = 'c', long = "config")]
    config: Option<String>,

    #[arg(long)]
    listener: Option<String>,

    #[arg(long)]
    protocol: Option<String>,

    #[arg(long)]
    json: bool,

    #[arg(long, value_name = "URL")]
    admin: Option<String>,
}

#[cfg(feature = "pproxy-compat")]
#[derive(Parser, Debug)]
struct PproxyCommand {
    #[command(subcommand)]
    action: PproxyAction,
}

#[cfg(feature = "pproxy-compat")]
#[derive(Subcommand, Debug)]
enum PproxyAction {
    /// Translate pproxy arguments to Eggress TOML
    Translate(PproxyTranslate),
    /// Check pproxy arguments and report parity tier
    Check(PproxyCheck),
    /// Translate and run pproxy-style arguments
    Run(PproxyRun),
}

#[cfg(feature = "pproxy-compat")]
#[derive(Parser, Debug)]
struct PproxyTranslate {
    /// pproxy-style arguments (after --)
    #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
    args: Vec<String>,

    /// Add explanatory comments to generated TOML
    #[arg(long)]
    annotate: bool,
}

#[cfg(feature = "pproxy-compat")]
#[derive(Parser, Debug)]
struct PproxyCheck {
    /// pproxy-style arguments (after --)
    #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
    args: Vec<String>,

    /// Output results as JSON
    #[arg(long)]
    json: bool,
}

#[cfg(feature = "pproxy-compat")]
#[derive(Parser, Debug)]
struct PproxyRun {
    /// pproxy-style arguments (after --)
    #[arg(trailing_var_arg = true, allow_hyphen_values = true)]
    args: Vec<String>,

    #[arg(long = "log-format", value_name = "FORMAT", default_value = "pretty")]
    log_format: String,
}

fn init_logging(format: &str) {
    let builder = fmt().with_env_filter(
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")),
    );

    match format {
        "json" => builder.json().init(),
        "compact" => builder.compact().init(),
        "pretty" => builder.pretty().init(),
        _ => builder.pretty().init(),
    }
}

fn handle_route_explain(args: &RouteExplain) {
    if let Some(ref admin_url) = args.admin {
        handle_route_explain_remote(args, admin_url);
        return;
    }

    let (router, is_online) = match &args.config {
        Some(path) => match eggress_config::compile::load_and_compile(path) {
            Ok(rt) => match build_router_from_config(&rt) {
                Ok(r) => (r, true),
                Err(e) => {
                    eprintln!("failed to build router from config: {e}");
                    std::process::exit(EXIT_CONFIG_VALIDATION);
                }
            },
            Err(e) => {
                eprintln!("failed to load config: {e}");
                std::process::exit(EXIT_CONFIG_VALIDATION);
            }
        },
        None => (Router::new(vec![], RouteActionSpec::Direct), false),
    };

    let target: eggress_core::TargetAddr = match args.target.parse() {
        Ok(t) => t,
        Err(e) => {
            eprintln!("{e}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    let protocol = match args.protocol.as_deref() {
        Some("http") => eggress_core::ProtocolId::Http,
        Some("socks4") => eggress_core::ProtocolId::Socks4,
        Some("socks5") => eggress_core::ProtocolId::Socks5,
        Some(p) => {
            eprintln!("unknown protocol: {p}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
        None => eggress_core::ProtocolId::Http,
    };

    let listener = args.listener.as_deref().unwrap_or("cli");

    let request = eggress_routing::RouteRequest {
        target: &target,
        source: None,
        listener,
        inbound_protocol: protocol,
        identity: &eggress_core::ClientIdentity::Anonymous,
        transport: eggress_routing::TransportKind::Tcp,
    };

    let explanation = router.explain(&request, 0);

    if args.json {
        match serde_json::to_string_pretty(&explanation) {
            Ok(json) => println!("{json}"),
            Err(e) => {
                eprintln!("failed to serialize explanation: {e}");
                std::process::exit(EXIT_RUNTIME_FAILURE);
            }
        }
    } else {
        print_explanation(&explanation, is_online);
    }
}

fn handle_route_explain_remote(args: &RouteExplain, admin_url: &str) {
    let target: eggress_core::TargetAddr = match args.target.parse() {
        Ok(t) => t,
        Err(e) => {
            eprintln!("{e}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    let protocol = match args.protocol.as_deref() {
        Some("http") => "http",
        Some("socks4") => "socks4",
        Some("socks5") => "socks5",
        Some(p) => {
            eprintln!("unknown protocol: {p}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
        None => "http",
    };

    let listener = args.listener.as_deref().unwrap_or("default");

    let body = serde_json::json!({
        "target": target.to_string(),
        "listener": listener,
        "protocol": protocol,
    });

    let base = admin_url.trim_end_matches('/');
    let url = format!("{base}/-/route-explain");

    let (host, port, path) = parse_admin_url(&url);
    let host_header = if host.contains(':') {
        format!("[{host}]:{port}")
    } else {
        format!("{host}:{port}")
    };

    let body_str = body.to_string();
    let request = format!(
        "POST {path} HTTP/1.1\r\nHost: {host_header}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body_str}",
        body_str.len(),
    );

    let result = eggress_cli::run_async_test(move || {
        let host = host.clone();
        let port = port;
        let request = request.clone();
        Box::pin(async move {
            let addr = if host.contains(':') {
                format!("[{host}]:{port}")
            } else {
                format!("{host}:{port}")
            };
            // Return errors instead of exiting the process here: the future
            // runs inside a tokio runtime and must not bypass destructors.
            let mut stream = match tokio::net::TcpStream::connect(&addr).await {
                Ok(s) => s,
                Err(e) => return Err(format!("failed to connect to admin at {addr}: {e}")),
            };

            use tokio::io::{AsyncReadExt, AsyncWriteExt};
            if let Err(e) = stream.write_all(request.as_bytes()).await {
                return Err(format!("failed to send request: {e}"));
            }
            let _ = stream.shutdown().await;

            let mut response = Vec::new();
            loop {
                let mut buf = [0u8; 4096];
                match stream.read(&mut buf).await {
                    Ok(0) => break,
                    Ok(n) => response.extend_from_slice(&buf[..n]),
                    Err(_) => break,
                }
            }
            Ok(String::from_utf8_lossy(&response).to_string())
        })
    });

    let result = match result {
        Ok(result) => result,
        Err(message) => {
            eprintln!("{message}");
            std::process::exit(EXIT_RUNTIME_FAILURE);
        }
    };

    let body_start = result.find("\r\n\r\n").map(|i| i + 4).unwrap_or(0);
    let body = &result[body_start..];

    let status_line = result.lines().next().unwrap_or("");
    let status = status_line
        .split_whitespace()
        .nth(1)
        .and_then(|s| s.parse::<u16>().ok())
        .unwrap_or(0);

    if status != 200 {
        eprintln!("admin returned {status}: {body}");
        std::process::exit(EXIT_RUNTIME_FAILURE);
    }

    if args.json {
        println!("{body}");
    } else {
        match serde_json::from_str::<eggress_routing::RouteExplanation>(body) {
            Ok(explanation) => print_explanation(&explanation, true),
            Err(e) => {
                eprintln!("failed to parse response: {e}");
                std::process::exit(EXIT_RUNTIME_FAILURE);
            }
        }
    }
}

fn parse_admin_url(url: &str) -> (String, u16, String) {
    let without_proto = url
        .strip_prefix("http://")
        .or_else(|| url.strip_prefix("https://"))
        .unwrap_or(url);
    let (host_port, path) = match without_proto.find('/') {
        Some(i) => (&without_proto[..i], &without_proto[i..]),
        None => (without_proto, "/"),
    };
    let (host, port) = if let Some(rest) = host_port.strip_prefix('[') {
        let close = rest.find(']').unwrap_or(rest.len());
        let host = rest[..close].to_string();
        let after = rest[close..].strip_prefix(']').unwrap_or("");
        let port = after
            .strip_prefix(':')
            .and_then(|p| p.parse::<u16>().ok())
            .unwrap_or(9090);
        (host, port)
    } else {
        match host_port.rfind(':') {
            Some(i) => (
                host_port[..i].to_string(),
                host_port[i + 1..].parse::<u16>().unwrap_or(9090),
            ),
            None => (host_port.to_string(), 9090),
        }
    };
    (host, port, path.to_string())
}

fn handle_upstream_test(args: &UpstreamTest) {
    let rt = match &args.config {
        Some(path) => match eggress_config::compile::load_and_compile(path) {
            Ok(rt) => rt,
            Err(e) => {
                eprintln!("failed to load config: {e}");
                std::process::exit(EXIT_CONFIG_VALIDATION);
            }
        },
        None => {
            eprintln!("--config is required for upstream test");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    // Filter upstreams by --id if provided
    let rt = if let Some(ref id) = args.id {
        let mut filtered = rt;
        filtered.upstreams.retain(|u| &u.id == id);
        filtered
    } else {
        rt
    };

    if rt.upstreams.is_empty() {
        eprintln!("no upstreams found matching criteria");
        std::process::exit(EXIT_CONFIG_VALIDATION);
    }

    let timeout = Duration::from_secs(args.timeout);
    let exit_code = eggress_cli::run_upstream_test_with_mode(
        &rt,
        args.target.as_deref(),
        &args.mode,
        timeout,
        args.json,
    );
    std::process::exit(exit_code);
}

#[cfg(feature = "pproxy-compat")]
fn handle_pproxy_translate(args: &PproxyTranslate) {
    let pproxy_args = match eggress_pproxy_compat::PproxyArgs::parse(&args.args) {
        Ok(a) => a,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    let output = match eggress_pproxy_compat::translate_pproxy_args(&pproxy_args) {
        Ok(o) => o,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CONFIG_VALIDATION);
        }
    };

    if output.has_unsupported() {
        for u in &output.unsupported {
            eprintln!("warning: {u}");
        }
        eprintln!("\nGenerated TOML may be incomplete due to unsupported features.");
    }

    for w in &output.warnings {
        eprintln!("warning: {w}");
    }

    if args.annotate {
        println!("# Generated by eggress pproxy translate");
        println!("# pproxy arguments: {}", args.args.join(" "));
        if !output.warnings.is_empty() || !output.unsupported.is_empty() {
            println!("#");
            for w in &output.warnings {
                println!("# {w}");
            }
            for u in &output.unsupported {
                println!("# {u}");
            }
        }
        println!();
    }

    print!("{}", output.toml);

    if output.has_unsupported() {
        std::process::exit(EXIT_UNSUPPORTED_FEATURE);
    }
}

#[cfg(feature = "pproxy-compat")]
fn handle_pproxy_check(args: &PproxyCheck) {
    let pproxy_args = match eggress_pproxy_compat::PproxyArgs::parse(&args.args) {
        Ok(a) => a,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    let output = match eggress_pproxy_compat::translate_pproxy_args(&pproxy_args) {
        Ok(o) => o,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CONFIG_VALIDATION);
        }
    };

    let local_uris = pproxy_args.parse_local_uris();
    let remote_chains = pproxy_args.parse_remote_chains();

    if args.json {
        let listeners = match &local_uris {
            Ok(uris) => uris
                .iter()
                .map(|u| u.redacted_display().to_string())
                .collect(),
            Err(e) => vec![format!("error: {e}")],
        };
        let remotes = match &remote_chains {
            Ok(chains) => chains.iter().map(|c| c.redacted_display()).collect(),
            Err(e) => vec![format!("error: {e}")],
        };
        let remote_chains_info: Vec<ChainInfo> = match &remote_chains {
            Ok(chains) => chains
                .iter()
                .map(|c| ChainInfo {
                    raw: c.raw.clone(),
                    hops: c.hops.len(),
                    hop_schemes: c.hops.iter().map(|h| h.scheme.clone()).collect(),
                    is_chain: c.hops.len() > 1,
                })
                .collect(),
            Err(e) => vec![ChainInfo {
                raw: format!("error: {e}"),
                hops: 0,
                hop_schemes: vec![],
                is_chain: false,
            }],
        };

        let mut diagnostics: Vec<eggress_pproxy_compat::StructuredDiagnostic> = Vec::new();
        let mut features: Vec<FeatureInfo> = Vec::new();

        for w in &output.warnings {
            let feature_tier = eggress_pproxy_compat::manifest_tier_for_category(w.category);
            diagnostics.push(eggress_pproxy_compat::StructuredDiagnostic::from(w));
            features.push(FeatureInfo {
                name: w.category.to_string(),
                tier: feature_tier.as_str().to_string(),
                diagnostic_code: Some(w.diagnostic_code()),
            });
        }
        for u in &output.unsupported {
            let diag = eggress_pproxy_compat::StructuredDiagnostic {
                code: eggress_pproxy_compat::DiagnosticCode::UnsupportedProtocol,
                feature_id: Some(u.feature.to_string()),
                tier: Some(
                    eggress_pproxy_compat::ManifestTier::Unsupported
                        .as_str()
                        .to_string(),
                ),
                message: u.detail.clone(),
                suggestion: None,
            };
            diagnostics.push(diag);
            features.push(FeatureInfo {
                name: u.feature.to_string(),
                tier: eggress_pproxy_compat::ManifestTier::Unsupported
                    .as_str()
                    .to_string(),
                diagnostic_code: Some(eggress_pproxy_compat::DiagnosticCode::UnsupportedProtocol),
            });
        }

        let tier =
            eggress_pproxy_compat::classify_aggregate_tier(&output.warnings, &output.unsupported);

        let check_output = PproxyCheckOutput {
            tier: tier.as_str().to_string(),
            diagnostics,
            features,
            raw_args: args.args.clone(),
            parsed_uris: ParsedUris {
                listeners,
                remotes,
                chain_info: remote_chains_info,
            },
        };

        match serde_json::to_string_pretty(&check_output) {
            Ok(json) => println!("{json}"),
            Err(e) => {
                eprintln!("failed to serialize check output: {e}");
                std::process::exit(EXIT_RUNTIME_FAILURE);
            }
        }
    } else {
        println!("pproxy compatibility check");
        println!("=========================");

        match local_uris {
            Ok(uris) => {
                for uri in &uris {
                    println!(
                        "  local:  {} -> scheme={}",
                        uri.redacted_display(),
                        uri.scheme
                    );
                }
            }
            Err(e) => eprintln!("  local:  error: {e}"),
        }

        match remote_chains {
            Ok(chains) => {
                for chain in &chains {
                    if chain.hops.len() > 1 {
                        println!(
                            "  remote: {} -> chain ({} hops: {})",
                            chain.redacted_display(),
                            chain.hops.len(),
                            chain
                                .hops
                                .iter()
                                .map(|h| h.scheme.as_str())
                                .collect::<Vec<_>>()
                                .join(" -> ")
                        );
                    } else if let Some(hop) = chain.hops.first() {
                        println!(
                            "  remote: {} -> scheme={}",
                            hop.redacted_display(),
                            hop.scheme
                        );
                    }
                }
            }
            Err(e) => eprintln!("  remote: error: {e}"),
        }

        let tier =
            eggress_pproxy_compat::classify_aggregate_tier(&output.warnings, &output.unsupported);
        println!("\nparity tier: {}", tier_label(&tier));

        if !output.warnings.is_empty() {
            println!("\nwarnings:");
            for w in &output.warnings {
                println!("  {w}");
            }
        }

        if !output.unsupported.is_empty() {
            println!("\nunsupported:");
            for u in &output.unsupported {
                println!("  {u}");
            }
        }
    }
}

#[cfg(feature = "pproxy-compat")]
fn handle_pproxy_run(args: &PproxyRun) {
    let pproxy_args = match if args.args.is_empty() {
        Ok(eggress_pproxy_compat::PproxyArgs::default_args())
    } else {
        eggress_pproxy_compat::PproxyArgs::parse(&args.args)
    } {
        Ok(a) => a,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
    };

    if pproxy_args.version {
        println!("{}", eggress_pproxy_compat::PproxyArgs::version_string());
        return;
    }
    if pproxy_args.help {
        println!("pproxy compatibility binary (eggress-pproxy-compat)");
        println!("Use -l and -r with pproxy-compatible URIs.");
        println!("Run standalone `pproxy --help` for the complete option reference.");
        return;
    }

    if let Some(flag) = pproxy_args.strict_parser_violations().first() {
        eprintln!("error: unknown option or positional argument '{flag}'");
        std::process::exit(EXIT_CLI_PARSE_ERROR);
    }

    if let Err(e) = pproxy_args.validate_strict_values() {
        eprintln!("error: {e}");
        std::process::exit(EXIT_CLI_PARSE_ERROR);
    }

    let output = match eggress_pproxy_compat::translate_pproxy_args(&pproxy_args) {
        Ok(o) => o,
        Err(e) => {
            eprintln!("error: {e}");
            std::process::exit(EXIT_CONFIG_VALIDATION);
        }
    };

    // Fatal gating: unknown flags and unsupported features stop startup.
    // The shared gate is the single source of truth for the fail-closed
    // policy applied by every compatibility execution entry point.
    let gate = eggress_pproxy_compat::evaluate_execution_gate(&pproxy_args, &output);
    if !gate.allows_start() {
        for blocker in &gate.blockers {
            match blocker {
                eggress_pproxy_compat::BlockReason::UnknownFlag(flag) => {
                    eprintln!("error: unknown option '{flag}'");
                }
                eggress_pproxy_compat::BlockReason::Unsupported(u) => {
                    eprintln!("warning: {u}");
                }
            }
        }
        eprintln!();
        if gate
            .blockers
            .iter()
            .any(|b| matches!(b, eggress_pproxy_compat::BlockReason::UnknownFlag(_)))
        {
            eprintln!("Run 'eggress pproxy check -- <args>' for supported options.");
            std::process::exit(EXIT_CLI_PARSE_ERROR);
        }
        eprintln!("Some features are unsupported. Service may not behave as expected.");
        eprintln!("refusing to start: unsupported features would yield a non-functional proxy.");
        std::process::exit(EXIT_UNSUPPORTED_FEATURE);
    }

    for w in &gate.warnings {
        eprintln!("warning: {w}");
    }

    // Compatibility verbosity/debug defaults come from the parsed pproxy
    // flags. An explicit RUST_LOG remains authoritative.
    init_pproxy_logging(&pproxy_args, &args.log_format);

    let test_target = pproxy_args.test_target();

    // Parse translated TOML into a validated RuntimeConfig in-memory.
    // No temporary file is created; the config lives entirely in process memory.
    let (rt_config, _warnings) =
        match eggress_config::validate_and_compile_toml_with_warnings(&output.toml) {
            Ok(r) => r,
            Err(e) => {
                eprintln!("config error: {e}");
                std::process::exit(EXIT_CONFIG_VALIDATION);
            }
        };

    if let Some(target) = test_target {
        let timeout = Duration::from_secs(10);
        let target = match eggress_cli::parse_pproxy_test_target(target) {
            Ok(target) => target.to_string(),
            Err(error) => {
                eprintln!("error: {error}");
                std::process::exit(EXIT_CLI_PARSE_ERROR);
            }
        };
        if rt_config.upstreams.is_empty() {
            std::process::exit(EXIT_SUCCESS);
        }
        let exit_code = eggress_cli::run_upstream_test(&rt_config, Some(&target), timeout, false);
        std::process::exit(exit_code);
    }

    #[cfg(feature = "pproxy-daemon")]
    if let Err(error) = eggress_cli::maybe_daemonize(pproxy_args.daemon) {
        eprintln!("error: {error}");
        std::process::exit(EXIT_UNSUPPORTED_FEATURE);
    }

    tracing::info!("starting eggress with pproxy-compatible config");

    // Start from the in-memory RuntimeConfig. No config file path is provided,
    // so SIGHUP reload is disabled (there is no stable user-authored config
    // file to reload from in compatibility mode).
    let compatibility_options = eggress_runtime::CompatibilityOptions {
        compatibility_mode: true,
        auth_timeout: Some(pproxy_args.effective_auth_timeout()),
        system_proxy: pproxy_args.system_proxy,
        debug: pproxy_args.debug,
        verbose_level: pproxy_args.verbose_level,
    };
    match eggress_runtime::ServiceSupervisor::start_from_config_with_options(
        rt_config,
        None,
        compatibility_options,
    ) {
        Ok(mut supervisor) => {
            if let Err(e) = supervisor.run() {
                eprintln!("runtime error: {e}");
                std::process::exit(EXIT_RUNTIME_FAILURE);
            }
        }
        Err(e) => {
            eprintln!("runtime error: {e}");
            std::process::exit(EXIT_RUNTIME_FAILURE);
        }
    }
}

#[cfg(feature = "pproxy-compat")]
fn init_pproxy_logging(pproxy_args: &eggress_pproxy_compat::PproxyArgs, format: &str) {
    let builder = fmt().with_env_filter(
        EnvFilter::try_from_default_env()
            .unwrap_or_else(|_| EnvFilter::new(pproxy_args.default_log_level())),
    );

    match format {
        "json" => builder.json().init(),
        "compact" => builder.compact().init(),
        _ => builder.compact().init(),
    }
}

#[cfg(feature = "pproxy-compat")]
#[derive(serde::Serialize)]
struct PproxyCheckOutput {
    tier: String,
    diagnostics: Vec<eggress_pproxy_compat::StructuredDiagnostic>,
    features: Vec<FeatureInfo>,
    raw_args: Vec<String>,
    parsed_uris: ParsedUris,
}

#[cfg(feature = "pproxy-compat")]
#[derive(serde::Serialize)]
struct FeatureInfo {
    name: String,
    tier: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    diagnostic_code: Option<eggress_pproxy_compat::DiagnosticCode>,
}

#[cfg(feature = "pproxy-compat")]
#[derive(serde::Serialize)]
struct ParsedUris {
    listeners: Vec<String>,
    remotes: Vec<String>,
    chain_info: Vec<ChainInfo>,
}

#[cfg(feature = "pproxy-compat")]
#[derive(serde::Serialize)]
struct ChainInfo {
    raw: String,
    hops: usize,
    hop_schemes: Vec<String>,
    is_chain: bool,
}

#[cfg(feature = "pproxy-compat")]
fn tier_label(tier: &eggress_pproxy_compat::ManifestTier) -> &'static str {
    match tier {
        eggress_pproxy_compat::ManifestTier::DropIn => "Drop-in",
        eggress_pproxy_compat::ManifestTier::CompatibleWithWarning => "Compatible (with warnings)",
        eggress_pproxy_compat::ManifestTier::NativeEquivalent => "Native equivalent",
        eggress_pproxy_compat::ManifestTier::IntentionalNonParity => "Intentional non-parity",
        eggress_pproxy_compat::ManifestTier::Unsupported => "Unsupported",
    }
}

#[cfg(feature = "operations")]
fn handle_system_proxy_inspect(args: &SystemProxyInspect) {
    let result = eggress_system_proxy::inspect_system_proxy();

    if args.json {
        match serde_json::to_string_pretty(&result) {
            Ok(json) => println!("{json}"),
            Err(e) => {
                eprintln!("failed to serialize inspection result: {e}");
                std::process::exit(EXIT_RUNTIME_FAILURE);
            }
        }
    } else {
        print_inspection_result(&result);
    }
}

#[cfg(feature = "operations")]
fn print_inspection_result(result: &eggress_system_proxy::InspectionResult) {
    println!("System Proxy Inspection");
    println!("=======================");
    println!("Platform: {}", result.platform);
    println!();

    println!("Capabilities:");
    for cap in &result.capabilities {
        println!("  {cap}");
    }
    println!();

    if let Some(ref settings) = result.settings {
        println!("Current Settings (source: {}):", settings.source);
        if let Some(ref http) = settings.http_proxy {
            println!("  HTTP proxy:  {http}");
        }
        if let Some(ref https) = settings.https_proxy {
            println!("  HTTPS proxy: {https}");
        }
        if let Some(ref socks) = settings.socks_proxy {
            println!("  SOCKS proxy: {socks}");
        }
        if let Some(ref no_proxy) = settings.no_proxy {
            println!("  No proxy:    {no_proxy}");
        }
    } else {
        println!("No proxy settings detected.");
    }
    println!();

    println!("Apply supported: {}", result.apply_supported);

    if !result.dry_run_commands.is_empty() {
        println!();
        println!("Dry-run apply commands:");
        for cmd in &result.dry_run_commands {
            println!("  {cmd}");
        }
    }
}

fn print_explanation(explanation: &eggress_routing::RouteExplanation, is_online: bool) {
    println!("Target: {}", explanation.target);
    println!("Listener: {}", explanation.listener);
    println!("Protocol: {}", explanation.protocol);
    if let Some(ref rule) = explanation.matched_rule {
        println!("Matched rule: {rule}");
    }
    println!("Action: {}", explanation.action);
    if let Some(ref group) = explanation.upstream_group {
        println!("Upstream group: {group}");
    }
    if let Some(ref scheduler) = explanation.scheduler {
        println!("Scheduler: {scheduler}");
    }
    if !explanation.eligible_upstreams.is_empty() {
        println!("Eligible upstreams:");
        for u in &explanation.eligible_upstreams {
            println!(
                "  {}  {}  active={}  in_flight={}",
                u.id, u.health, u.active, u.in_flight
            );
        }
    }
    if let Some(ref upstream) = explanation.selected_upstream {
        println!("Selected upstream: {upstream}");
    }
    if let Some(ref chain) = explanation.chain {
        println!("Chain: {chain}");
    }
    if is_online {
        println!("Config generation: {}", explanation.generation);
    } else {
        println!("Mode: offline");
        println!("Generation: not-live");
    }
}

fn build_router_from_cli(args: &Cli) -> Result<Router, Box<dyn std::error::Error + Send + Sync>> {
    let upstream_chain: Option<eggress_uri::ProxyChainSpec> = if args.upstreams.is_empty() {
        None
    } else {
        let combined = args.upstreams.join("__");
        match eggress_uri::parse_proxy_chain(&combined) {
            Ok(spec) => Some(spec),
            Err(e) => return Err(format!("invalid upstream URI: {e}").into()),
        }
    };

    let mut rules: Vec<eggress_routing::CompiledRule> = Vec::new();
    let mut default_action = RouteActionSpec::Direct;
    let mut groups = Vec::new();

    if let Some(ref spec) = upstream_chain {
        let upstream = Arc::new(eggress_routing::upstream::UpstreamRuntime::new(
            eggress_core::UpstreamId::new("cli-upstream"),
            spec.clone(),
        ));
        let group_id = eggress_routing::UpstreamGroupId(Arc::from("cli-group"));
        let group = eggress_routing::upstream::UpstreamGroup::new(
            group_id.clone(),
            eggress_routing::scheduler::SchedulerKind::FirstAvailable,
            Arc::from([upstream]),
            eggress_routing::upstream::GroupFallback::Direct,
        );
        default_action = RouteActionSpec::UpstreamGroup(group_id.clone());
        groups.push((group_id, group));
    }

    if let Some(ref rules_file_path) = args.rules_file {
        let content = std::fs::read_to_string(rules_file_path)
            .map_err(|e| format!("failed to read rules file '{}': {}", rules_file_path, e))?;
        let compat_rules = eggress_routing::CompatRegexRule::parse_file(&content)
            .map_err(|e| format!("failed to parse rules file '{}': {}", rules_file_path, e))?;
        for (idx, compat) in compat_rules.into_iter().enumerate() {
            rules.push(eggress_routing::CompiledRule {
                id: eggress_routing::RuleId(Arc::from(format!("rules-file-{}", idx + 1).as_str())),
                matcher: eggress_routing::MatchExpr::HostRegex(compat.pattern),
                action: default_action.clone(),
            });
        }
    }

    Ok(Router::with_groups(rules, default_action, groups))
}

struct ListenerSpec {
    bind_addr: SocketAddr,
    protocols: Vec<eggress_core::ProtocolId>,
    auth: eggress_server::accept::InboundAuthentication,
}

fn parse_listener_uri(uri: &str) -> Result<ListenerSpec, Box<dyn std::error::Error + Send + Sync>> {
    let spec = eggress_uri::parse_proxy_chain(uri)?;
    let first_hop = &spec.hops[0];
    let bind_addr: SocketAddr =
        format!("{}:{}", first_hop.endpoint.host, first_hop.endpoint.port).parse()?;

    let mut protocols: Vec<eggress_core::ProtocolId> =
        Vec::with_capacity(first_hop.protocols.len());
    for p in &first_hop.protocols {
        let id = match p {
            eggress_uri::ProtocolSpec::Http => eggress_core::ProtocolId::Http,
            eggress_uri::ProtocolSpec::HttpOnly => eggress_core::ProtocolId::Http,
            eggress_uri::ProtocolSpec::Socks4 => eggress_core::ProtocolId::Socks4,
            eggress_uri::ProtocolSpec::Socks5 => eggress_core::ProtocolId::Socks5,
            eggress_uri::ProtocolSpec::Shadowsocks => eggress_core::ProtocolId::Shadowsocks,
            eggress_uri::ProtocolSpec::ShadowsocksR => eggress_core::ProtocolId::ShadowsocksR,
            eggress_uri::ProtocolSpec::Trojan => eggress_core::ProtocolId::Trojan,
            eggress_uri::ProtocolSpec::Http2 => eggress_core::ProtocolId::Http2,
            eggress_uri::ProtocolSpec::Http3 => eggress_core::ProtocolId::Http3,
            eggress_uri::ProtocolSpec::Quic => eggress_core::ProtocolId::Quic,
            eggress_uri::ProtocolSpec::WebSocket => eggress_core::ProtocolId::WebSocket,
            eggress_uri::ProtocolSpec::Raw => eggress_core::ProtocolId::Raw,
            eggress_uri::ProtocolSpec::Unix => eggress_core::ProtocolId::Raw,
            eggress_uri::ProtocolSpec::Ssh => {
                return Err("SSH is an upstream-only transport, not a listener protocol".into())
            }
        };
        protocols.push(id);
    }

    let auth = match &first_hop.credentials {
        Some(credentials) => eggress_server::accept::InboundAuthentication::UsernamePassword {
            username: credentials.username.clone(),
            password: credentials.password.clone(),
        },
        None => eggress_server::accept::InboundAuthentication::None,
    };

    Ok(ListenerSpec {
        bind_addr,
        protocols,
        auth,
    })
}

#[tokio::main]
async fn main() -> ExitCode {
    let exit_code = run().await;
    let code = u8::try_from(exit_code).unwrap_or_else(|_| {
        debug_assert!(
            (0..=u8::MAX as i32).contains(&exit_code),
            "exit code {exit_code} does not fit in u8"
        );
        EXIT_RUNTIME_FAILURE as u8
    });
    ExitCode::from(code)
}

async fn run() -> i32 {
    let args = Cli::parse();

    if let Some(SubCommand::Route(explain_args)) = args.command {
        handle_route_explain(&explain_args);
        return EXIT_SUCCESS;
    }

    if let Some(SubCommand::Upstream(upstream_cmd)) = args.command {
        match upstream_cmd.action {
            UpstreamAction::Test(test_args) => {
                handle_upstream_test(&test_args);
                return EXIT_SUCCESS;
            }
        }
    }

    #[cfg(feature = "pproxy-compat")]
    if let Some(SubCommand::Pproxy(pproxy_cmd)) = args.command {
        match pproxy_cmd.action {
            PproxyAction::Translate(translate_args) => {
                handle_pproxy_translate(&translate_args);
                return EXIT_SUCCESS;
            }
            PproxyAction::Check(check_args) => {
                handle_pproxy_check(&check_args);
                return EXIT_SUCCESS;
            }
            PproxyAction::Run(run_args) => {
                handle_pproxy_run(&run_args);
                return EXIT_SUCCESS;
            }
        }
    }

    #[cfg(feature = "operations")]
    if let Some(SubCommand::SystemProxy(sysproxy_cmd)) = args.command {
        match sysproxy_cmd.action {
            SystemProxyAction::Inspect(inspect_args) => {
                handle_system_proxy_inspect(&inspect_args);
                return EXIT_SUCCESS;
            }
        }
    }

    if args.config.is_some() && (!args.listeners.is_empty() || !args.upstreams.is_empty()) {
        eprintln!("--config mode is incompatible with -l and -r flags. Use one or the other.");
        return EXIT_CLI_PARSE_ERROR;
    }

    if let Some(ref config_path) = args.config {
        init_logging(&args.log_format);
        match eggress_runtime::ServiceSupervisor::start(config_path) {
            Ok(mut supervisor) => {
                if let Err(e) = supervisor.run() {
                    eprintln!("runtime error: {e}");
                    return EXIT_RUNTIME_FAILURE;
                }
            }
            Err(e) => {
                eprintln!("runtime error: {e}");
                return EXIT_RUNTIME_FAILURE;
            }
        }
        return EXIT_SUCCESS;
    }

    init_logging(&args.log_format);
    let cancel_token = CancellationToken::new();

    let router = match build_router_from_cli(&args) {
        Ok(r) => r,
        Err(e) => {
            eprintln!("{e}");
            return EXIT_CONFIG_VALIDATION;
        }
    };

    let routing_service = Arc::new(SharedRoutingService::new(router));

    let metrics = Arc::new(eggress_metrics::MetricsRegistry::new());

    let listener_uris: Vec<String> = if args.listeners.is_empty() {
        vec!["http://127.0.0.1:8080".to_string()]
    } else {
        args.listeners
    };

    let mut listener_specs = Vec::new();
    for uri in &listener_uris {
        match parse_listener_uri(uri) {
            Ok(spec) => listener_specs.push((uri.clone(), spec)),
            Err(e) => {
                eprintln!(
                    "invalid listener URI '{}': {e}",
                    eggress_uri::redact_proxy_uri(uri)
                );
                return EXIT_CLI_PARSE_ERROR;
            }
        }
    }

    let mut handles = Vec::new();

    for (uri, spec) in &listener_specs {
        let cancel = cancel_token.clone();
        let routing = routing_service.clone();
        let metrics = metrics.clone();
        let bind_addr = spec.bind_addr;
        let protocols = spec.protocols.clone();
        let auth = spec.auth.clone();
        let uri = uri.clone();

        let handle = tokio::spawn(async move {
            if let Err(e) = run_listener(bind_addr, protocols, routing, auth, metrics, cancel).await
            {
                tracing::error!(
                    "listener '{}' error: {e}",
                    eggress_uri::redact_proxy_uri(&uri)
                );
            }
        });
        handles.push(handle);
    }

    tracing::info!("eggress started, {} listener(s)", listener_specs.len());

    let mut shutdown_handles = handles;

    let shutdown_exit_code: i32;

    {
        let token = cancel_token.clone();

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

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

            let exit;
            loop {
                tokio::select! {
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("shutdown signal received");
                        token.cancel();
                        exit = EXIT_SIGINT;
                        break;
                    }
                    _ = async { sigterm.as_mut().ok()?.recv().await }, if sigterm.is_ok() => {
                        tracing::info!("shutdown signal received");
                        token.cancel();
                        exit = EXIT_SIGTERM;
                        break;
                    }
                    _ = async { sighup.as_mut().ok()?.recv().await }, if sighup.is_ok() => {
                        tracing::warn!("SIGHUP received but no config file specified in compatibility mode, ignoring");
                    }
                }
            }
            shutdown_exit_code = exit;
        }

        #[cfg(not(unix))]
        {
            tokio::signal::ctrl_c().await.ok();
            tracing::info!("shutdown signal received");
            token.cancel();
            shutdown_exit_code = EXIT_SIGINT;
        }
    }

    tracing::info!("draining active connections");

    let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
    loop {
        let notified = ACTIVE_CONNECTIONS_DRAIN.notified();
        let active = ACTIVE_CONNECTIONS.load(Ordering::Relaxed);
        if active == 0 {
            tracing::info!("all connections drained");
            break;
        }
        if tokio::time::Instant::now() >= deadline {
            tracing::warn!(active, "drain timeout reached, forcing shutdown");
            break;
        }
        // Wake immediately when a connection completes so the drain doesn't
        // have to wait out a full tick to notice progress.
        notified.await;
    }

    for h in shutdown_handles.drain(..) {
        let _ = h.await;
    }

    tracing::info!("eggress stopped");
    shutdown_exit_code
}

fn build_router_from_config(
    rt: &eggress_config::RuntimeConfig,
) -> Result<Router, Box<dyn std::error::Error + Send + Sync>> {
    let mut seen_upstream_ids = std::collections::HashSet::new();
    let mut upstreams = Vec::new();

    for u in &rt.upstreams {
        if !seen_upstream_ids.insert(u.id.clone()) {
            return Err(format!("duplicate upstream ID '{}'", u.id).into());
        }
        let id = eggress_routing::UpstreamGroupId(Arc::from(u.id.as_str()));
        let runtime = eggress_routing::upstream::UpstreamRuntime::new(
            eggress_core::UpstreamId::new(u.id.clone()),
            u.chain.clone(),
        );
        upstreams.push((id, runtime));
    }

    let upstream_map: std::collections::HashMap<
        String,
        Arc<eggress_routing::upstream::UpstreamRuntime>,
    > = upstreams
        .into_iter()
        .map(|(id, runtime)| (id.0.to_string(), Arc::new(runtime)))
        .collect();

    let mut seen_group_ids = std::collections::HashSet::new();
    let mut groups = Vec::new();

    for g in &rt.groups {
        if !seen_group_ids.insert(g.id.clone()) {
            return Err(format!("duplicate group ID '{}'", g.id).into());
        }
        let mut members = Vec::new();
        for m in &g.members {
            let member = upstream_map
                .get(m)
                .ok_or_else(|| format!("group '{}' references unknown upstream '{}'", g.id, m))?;
            members.push(member.clone());
        }
        if members.is_empty() {
            return Err(format!("group '{}' has no valid members", g.id).into());
        }

        let fallback = match g.fallback {
            eggress_config::compile::GroupFallback::Reject => {
                eggress_routing::upstream::GroupFallback::Reject
            }
            eggress_config::compile::GroupFallback::Direct => {
                eggress_routing::upstream::GroupFallback::Direct
            }
            eggress_config::compile::GroupFallback::UseUnhealthy => {
                eggress_routing::upstream::GroupFallback::UseUnhealthy
            }
        };

        groups.push((
            g.id.clone(),
            eggress_routing::upstream::UpstreamGroup::new(
                g.id.clone(),
                g.scheduler,
                Arc::from(members),
                fallback,
            ),
        ));
    }

    let group_ids: std::collections::HashSet<_> = groups.iter().map(|(id, _)| id.clone()).collect();

    let mut rules = Vec::new();
    for r in &rt.rules {
        let action = match &r.action {
            eggress_routing::RouteActionSpec::Direct => eggress_routing::RouteActionSpec::Direct,
            eggress_routing::RouteActionSpec::UpstreamGroup(gid) => {
                if !group_ids.contains(gid) {
                    return Err(
                        format!("rule '{}' references unknown group '{}'", r.id, gid).into(),
                    );
                }
                eggress_routing::RouteActionSpec::UpstreamGroup(gid.clone())
            }
            eggress_routing::RouteActionSpec::Reject(reason) => {
                eggress_routing::RouteActionSpec::Reject(reason.clone())
            }
        };
        rules.push(eggress_routing::CompiledRule {
            id: r.id.clone(),
            matcher: r.matcher.clone(),
            action,
        });
    }

    Ok(Router::with_groups(
        rules,
        rt.default_action.clone(),
        groups,
    ))
}

async fn run_listener(
    bind_addr: SocketAddr,
    protocols: Vec<eggress_core::ProtocolId>,
    routing: Arc<SharedRoutingService>,
    authentication: eggress_server::accept::InboundAuthentication,
    metrics: Arc<dyn eggress_server::SessionMetrics>,
    cancel_token: CancellationToken,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let config = TcpListenerConfig {
        bind_addr,
        protocols: protocols.clone(),
        auth_required: false,
        handshake_timeout: Duration::from_secs(30),
        connection_limit: 1024,
    };

    let listener = TcpListener::new(&config, cancel_token.clone()).await?;
    let local_addr = listener.local_addr()?;
    tracing::info!("listening on {local_addr}");

    let proto_slice: Arc<[eggress_core::ProtocolId]> = config.protocols.clone().into();

    // Back off when accept() keeps failing (e.g. fd exhaustion) so the loop
    // does not tight-spin while the system is resource-starved.
    const ACCEPT_ERROR_BACKOFF: Duration = Duration::from_millis(100);

    loop {
        let conn = match listener.accept().await {
            Ok(conn) => conn,
            Err(e) => {
                if eggress_core::listener::is_listener_cancelled(&e) {
                    break;
                }
                match e.kind() {
                    std::io::ErrorKind::WouldBlock
                    | std::io::ErrorKind::Interrupted
                    | std::io::ErrorKind::ConnectionAborted => {}
                    _ => tokio::time::sleep(ACCEPT_ERROR_BACKOFF).await,
                }
                tracing::error!("accept error: {e}");
                continue;
            }
        };

        let routing = routing.clone();
        let peer = conn.peer_addr;
        let listener = local_addr;
        let conn_id = CONNECTION_COUNTER.fetch_add(1, Ordering::Relaxed);
        let conn_protocols = proto_slice.clone();
        let conn_auth = authentication.clone();
        let conn_metrics = metrics.clone();

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

        tokio::spawn(async move {
            let started = std::time::Instant::now();
            let config = ConnectionConfig {
                routing: routing as Arc<dyn RouteService>,
                context: eggress_server::ConnectionContext {
                    source: Some(peer),
                    listener: listener.to_string(),
                    generation: 0,
                },
                handshake_timeout: Duration::from_secs(30),
                connect_timeout: Duration::from_secs(30),
                protocols: conn_protocols,
                authentication: conn_auth,
                metrics: Some(conn_metrics),
                udp: None,
                tls_client_config: None,
                shadowsocks: None,
                shadowsocks_metrics: None,
                trojan: None,
                fixed_target: None,
                local_bind: None,
                #[cfg(feature = "ssh")]
                ssh_sessions: None,
            };

            let report = eggress_server::serve_connection(conn.stream, config)
                .instrument(tracing::info_span!(
                    "conn",
                    id = conn_id,
                    peer = %peer,
                    listener = %listener,
                ))
                .await;

            ACTIVE_CONNECTIONS.fetch_sub(1, Ordering::Relaxed);
            ACTIVE_CONNECTIONS_DRAIN.notify_one();

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

    Ok(())
}

use tracing::Instrument;

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    #[tokio::test]
    async fn test_http_proxy_end_to_end() {
        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;

        let proxy_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let proxy_addr = proxy_listener.local_addr().unwrap();
        drop(proxy_listener);

        let cancel = CancellationToken::new();
        let config = TcpListenerConfig {
            bind_addr: proxy_addr,
            protocols: vec![eggress_core::ProtocolId::Http],
            auth_required: false,
            handshake_timeout: Duration::from_secs(5),
            connection_limit: 10,
        };
        let listener = TcpListener::new(&config, cancel.clone()).await.unwrap();

        let routing: Arc<SharedRoutingService> = Arc::new(SharedRoutingService::new(Router::new(
            vec![],
            RouteActionSpec::Direct,
        )));

        let proxy_jh = tokio::spawn(async move {
            loop {
                let conn = match listener.accept().await {
                    Ok(c) => c,
                    Err(_) => break,
                };
                let routing = routing.clone();
                let config = ConnectionConfig {
                    routing: routing as Arc<dyn RouteService>,
                    context: eggress_server::ConnectionContext {
                        source: Some(conn.peer_addr),
                        listener: String::new(),
                        generation: 0,
                    },
                    handshake_timeout: Duration::from_secs(5),
                    connect_timeout: Duration::from_secs(10),
                    protocols: Arc::from([eggress_core::ProtocolId::Http]),
                    authentication: eggress_server::accept::InboundAuthentication::None,
                    metrics: None,
                    udp: None,
                    tls_client_config: None,
                    shadowsocks: None,
                    shadowsocks_metrics: None,
                    trojan: None,
                    fixed_target: None,
                    local_bind: None,
                    #[cfg(feature = "ssh")]
                    ssh_sessions: None,
                };
                tokio::spawn(async move {
                    let _ = eggress_server::serve_connection(conn.stream, config).await;
                });
            }
        });

        let mut stream = tokio::net::TcpStream::connect(proxy_addr).await.unwrap();
        let connect_req = format!(
            "CONNECT {}:{} HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
            echo_addr.ip(),
            echo_addr.port(),
            echo_addr.ip(),
            echo_addr.port()
        );
        stream.write_all(connect_req.as_bytes()).await.unwrap();

        let mut response = vec![0u8; 1024];
        let n = stream.read(&mut response).await.unwrap();
        let response_str = String::from_utf8_lossy(&response[..n]);
        assert!(
            response_str.contains("200"),
            "expected 200, got: {response_str}"
        );

        let header_end = response_str.find("\r\n\r\n").unwrap() + 4;
        let leftover = &response.as_slice()[header_end..n];

        stream.write_all(b"hello proxy").await.unwrap();
        stream.shutdown().await.unwrap();

        let mut buf = Vec::new();
        if !leftover.is_empty() {
            buf.extend_from_slice(leftover);
        }
        stream.read_to_end(&mut buf).await.unwrap();
        assert_eq!(&buf, b"hello proxy");

        cancel.cancel();
        let _ = proxy_jh.await;
        echo_jh.abort();
    }

    #[tokio::test]
    async fn test_upstream_test_reachable() {
        let (echo_addr, echo_jh) = eggress_testkit::start_echo_server().await;

        let result = eggress_cli::test_upstream_tcp(
            &echo_addr.ip().to_string(),
            echo_addr.port(),
            Duration::from_secs(5),
        )
        .await;

        assert!(result.reachable);
        assert!(result.latency_ms.is_some());
        assert!(result.error.is_none());

        echo_jh.abort();
    }

    #[tokio::test]
    async fn test_upstream_test_unreachable() {
        let result = eggress_cli::test_upstream_tcp("127.0.0.1", 1, Duration::from_secs(1)).await;

        assert!(!result.reachable);
        assert!(result.latency_ms.is_none());
        assert!(result.error.is_some());
    }

    #[test]
    fn test_upstream_test_json_output() {
        let result = eggress_cli::UpstreamTestResult {
            id: "test-upstream".to_string(),
            host: "127.0.0.1".to_string(),
            port: 1080,
            target: "example.com:443".to_string(),
            mode: "tcp".to_string(),
            reachable: true,
            latency_ms: Some(15),
            error: None,
            failure: None,
            failed_hop: None,
        };

        let json = serde_json::to_string_pretty(&result).unwrap();
        assert!(json.contains("\"reachable\": true"));
        assert!(json.contains("\"latency_ms\": 15"));
        assert!(!json.contains("secret"));
    }

    #[test]
    fn parse_admin_url_default_port_ipv6_loopback() {
        let (host, port, path) = parse_admin_url("http://[::1]/-/route-explain");
        assert_eq!(host, "::1");
        assert_eq!(port, 9090);
        assert_eq!(path, "/-/route-explain");
    }

    #[test]
    fn parse_admin_url_explicit_port_ipv6_loopback() {
        let (host, port, path) = parse_admin_url("http://[::1]:9090/admin");
        assert_eq!(host, "::1");
        assert_eq!(port, 9090);
        assert_eq!(path, "/admin");
    }

    #[test]
    fn parse_admin_url_full_ipv6() {
        let (host, port, _path) = parse_admin_url("http://[2001:db8::1]:8080/-/route-explain");
        assert_eq!(host, "2001:db8::1");
        assert_eq!(port, 8080);
    }

    #[test]
    fn parse_admin_url_default_port_ipv4() {
        let (host, port, path) = parse_admin_url("http://127.0.0.1/admin");
        assert_eq!(host, "127.0.0.1");
        assert_eq!(port, 9090);
        assert_eq!(path, "/admin");
    }

    #[test]
    fn parse_admin_url_domain_with_port() {
        let (host, port, _path) = parse_admin_url("http://admin.example.com:8080/-/x");
        assert_eq!(host, "admin.example.com");
        assert_eq!(port, 8080);
    }

    #[test]
    fn listener_uri_redaction_hides_password() {
        // Regression for credential leak in listener parse/runtime error logs:
        // both sites must render through redact_proxy_uri.
        let uri = "socks5://secret_user:super_secret_password_123@127.0.0.1:1080";
        let redacted = eggress_uri::redact_proxy_uri(uri);
        assert!(
            !redacted.contains("super_secret_password_123"),
            "redacted listener URI leaked password: {redacted}"
        );
        assert!(
            !redacted.contains("secret_user"),
            "redacted listener URI leaked username: {redacted}"
        );
        assert!(redacted.contains("127.0.0.1:1080"));
    }
}