eggress-python 1.0.4

use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock};
use std::time::Duration;

static PY_CONNECTION_LIVE_COUNT: AtomicUsize = AtomicUsize::new(0);
static PY_CONNECTION_TOTAL_CREATED: AtomicUsize = AtomicUsize::new(0);
static PY_OUTBOUND_RUNTIME: OnceLock<Result<Arc<tokio::runtime::Runtime>, String>> =
    OnceLock::new();

fn outbound_runtime() -> Result<Arc<tokio::runtime::Runtime>, String> {
    PY_OUTBOUND_RUNTIME
        .get_or_init(|| {
            tokio::runtime::Builder::new_multi_thread()
                .enable_all()
                .build()
                .map(Arc::new)
                .map_err(|e| format!("runtime setup failed: {e}"))
        })
        .as_ref()
        .map(Arc::clone)
        .map_err(Clone::clone)
}

use pyo3::exceptions::{PyException, PyValueError};
use pyo3::prelude::*;
use pyo3::types::{PyDict, PyList, PyModule, PyModuleMethods, PySequence, PyString};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[pyclass]
struct PyAppliedSystemProxy {
    inner: Option<eggress_system_proxy::AppliedProxy>,
}

#[pymethods]
impl PyAppliedSystemProxy {
    fn restore(&mut self, py: Python<'_>) -> PyResult<()> {
        if let Some(mut applied) = self.inner.take() {
            py.detach(|| applied.restore())
                .map_err(PyValueError::new_err)?;
        }
        Ok(())
    }

    fn __enter__(slf: Py<Self>) -> Py<Self> {
        slf
    }

    fn __exit__(
        &mut self,
        py: Python<'_>,
        _exc_type: &Bound<'_, PyAny>,
        _exc_value: &Bound<'_, PyAny>,
        _traceback: &Bound<'_, PyAny>,
    ) -> PyResult<bool> {
        self.restore(py)?;
        Ok(false)
    }
}

#[pyfunction]
fn apply_system_proxy(py: Python<'_>, kind: &str, address: &str) -> PyResult<PyAppliedSystemProxy> {
    let kind = match kind {
        "http" => eggress_system_proxy::CompatibilityProxyKind::Http,
        "socks5" => eggress_system_proxy::CompatibilityProxyKind::Socks5,
        other => {
            return Err(PyValueError::new_err(format!(
                "unknown proxy kind: {other}"
            )))
        }
    };
    let address = address
        .parse()
        .map_err(|e| PyValueError::new_err(format!("invalid proxy address: {e}")))?;
    let inner = py
        .detach(|| eggress_system_proxy::apply_compatibility_proxy(kind, address))
        .map_err(PyValueError::new_err)?;
    Ok(PyAppliedSystemProxy { inner: Some(inner) })
}

pyo3::create_exception!(_eggress, EggressError, PyException);
pyo3::create_exception!(_eggress, ConfigError, EggressError);
pyo3::create_exception!(_eggress, StartupError, EggressError);
pyo3::create_exception!(_eggress, ReloadError, EggressError);
pyo3::create_exception!(_eggress, ShutdownError, EggressError);
pyo3::create_exception!(_eggress, UnsupportedFeatureError, EggressError);
pyo3::create_exception!(_eggress, InternalError, EggressError);
pyo3::create_exception!(_eggress, ConnectionError, EggressError);
pyo3::create_exception!(_eggress, ConnectionClosedError, EggressError);
pyo3::create_exception!(_eggress, TimeoutError, EggressError);
pyo3::create_exception!(_eggress, DnsError, EggressError);
pyo3::create_exception!(_eggress, AuthError, EggressError);
pyo3::create_exception!(_eggress, TlsError, EggressError);
pyo3::create_exception!(_eggress, LoopMismatchError, EggressError);
pyo3::create_exception!(_eggress, ConnectionCancelledError, EggressError);
pyo3::create_exception!(_eggress, UseAfterCloseError, EggressError);
pyo3::create_exception!(_eggress, UdpAssociationError, EggressError);
pyo3::create_exception!(_eggress, UnsupportedCompositionError, EggressError);

fn map_error(_py: Python<'_>, err: eggress_embed::EggressError) -> PyErr {
    use eggress_embed::EggressError as E;
    let msg = err.to_string();
    match err {
        E::Config(_) => ConfigError::new_err(msg),
        E::Runtime(_) => InternalError::new_err(msg),
        E::Startup(_) => StartupError::new_err(msg),
        E::Reload(_) => ReloadError::new_err(msg),
        E::Shutdown(_) => ShutdownError::new_err(msg),
        E::UnsupportedFeature { .. } => UnsupportedFeatureError::new_err(msg),
        E::Internal(_) => InternalError::new_err(msg),
    }
}

#[pyclass]
struct PyEggressConfig {
    inner: eggress_embed::EggressConfig,
}

#[pymethods]
impl PyEggressConfig {
    #[staticmethod]
    fn from_toml(py: Python<'_>, toml_str: &str) -> PyResult<Self> {
        let config = py
            .detach(|| eggress_embed::EggressConfig::from_toml_str(toml_str))
            .map_err(|e| map_error(py, e))?;
        Ok(Self { inner: config })
    }

    #[staticmethod]
    fn from_file(py: Python<'_>, path: &str) -> PyResult<Self> {
        let config = py
            .detach(|| eggress_embed::EggressConfig::from_toml_file(path))
            .map_err(|e| map_error(py, e))?;
        Ok(Self { inner: config })
    }

    fn redacted_toml(&self, py: Python<'_>) -> PyResult<String> {
        py.detach(|| self.inner.to_redacted_toml())
            .map_err(|e| map_error(py, e))
    }
}

#[pyclass]
struct PyEggressService {
    inner: Option<eggress_embed::EggressService>,
}

#[pymethods]
impl PyEggressService {
    #[new]
    fn new(_py: Python<'_>, config: &PyEggressConfig) -> Self {
        Self {
            inner: Some(eggress_embed::EggressService::new(config.inner.clone())),
        }
    }

    #[staticmethod]
    fn from_toml(py: Python<'_>, toml_str: &str) -> PyResult<Self> {
        let svc = py
            .detach(|| eggress_embed::EggressService::from_toml_str(toml_str))
            .map_err(|e| map_error(py, e))?;
        Ok(Self { inner: Some(svc) })
    }

    #[staticmethod]
    fn from_file(py: Python<'_>, path: &str) -> PyResult<Self> {
        let svc = py
            .detach(|| eggress_embed::EggressService::from_toml_file(path))
            .map_err(|e| map_error(py, e))?;
        Ok(Self { inner: Some(svc) })
    }

    fn start(&mut self, py: Python<'_>) -> PyResult<PyEggressHandle> {
        let svc = self
            .inner
            .take()
            .ok_or_else(|| EggressError::new_err("service already started"))?;
        let handle = py
            .detach(|| svc.start_blocking())
            .map_err(|e| map_error(py, e))?;
        Ok(PyEggressHandle {
            inner: Some(handle),
        })
    }

    /// Start with the compatibility-only runtime options parsed from pproxy
    /// arguments. Native Eggress service startup never uses this path.
    fn start_with_compatibility_options(
        &mut self,
        py: Python<'_>,
        auth_timeout_seconds: u64,
        system_proxy: bool,
        debug: bool,
        verbose_level: u8,
    ) -> PyResult<PyEggressHandle> {
        let svc = self
            .inner
            .take()
            .ok_or_else(|| EggressError::new_err("service already started"))?;
        let options = eggress_runtime::CompatibilityOptions {
            compatibility_mode: true,
            auth_timeout: Some(Duration::from_secs(auth_timeout_seconds)),
            system_proxy,
            debug,
            verbose_level,
        };
        let handle = py
            .detach(|| svc.start_blocking_with_compatibility_options(options))
            .map_err(|e| map_error(py, e))?;
        Ok(PyEggressHandle {
            inner: Some(handle),
        })
    }
}

#[pyclass]
struct PyEggressHandle {
    inner: Option<eggress_embed::EggressHandle>,
}

#[pymethods]
impl PyEggressHandle {
    fn bound_addresses(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
        let handle = self
            .inner
            .as_ref()
            .ok_or_else(|| EggressError::new_err("handle consumed"))?;
        let addrs = py.detach(|| handle.bound_addresses());
        let dict = PyDict::new(py);
        for la in &addrs.listeners {
            dict.set_item(&la.name, la.addr.to_string())?;
        }
        if let Some(admin) = addrs.admin {
            dict.set_item("_admin", admin.to_string())?;
        }
        Ok(dict.into())
    }

    fn status(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
        let handle = self
            .inner
            .as_ref()
            .ok_or_else(|| EggressError::new_err("handle consumed"))?;
        let st = py.detach(|| handle.status());
        let dict = PyDict::new(py);
        dict.set_item("generation", st.generation)?;
        dict.set_item("readiness", st.readiness)?;
        dict.set_item("active_connections", st.active_connections)?;
        dict.set_item("uptime_secs", st.uptime_secs)?;
        dict.set_item("listener_count", st.listener_count)?;
        dict.set_item("udp_associations_active", st.udp_associations_active)?;
        dict.set_item("upstream_count", st.upstream_count)?;
        let py_listeners = PyList::empty(py);
        for ls in &st.listeners {
            let ldict = PyDict::new(py);
            ldict.set_item("name", &ls.name)?;
            ldict.set_item("bind", &ls.bind)?;
            ldict.set_item("local_addr", ls.local_addr.to_string())?;
            ldict.set_item("protocols", &ls.protocols)?;
            ldict.set_item("udp_enabled", ls.udp_enabled)?;
            py_listeners.append(ldict)?;
        }
        dict.set_item("listeners", py_listeners)?;
        Ok(dict.into())
    }

    fn metrics_text(&self, py: Python<'_>) -> PyResult<String> {
        let handle = self
            .inner
            .as_ref()
            .ok_or_else(|| EggressError::new_err("handle consumed"))?;
        py.detach(|| handle.metrics_text())
            .map_err(|e| map_error(py, e))
    }

    fn reload_toml(&self, py: Python<'_>, toml_str: &str) -> PyResult<Py<PyDict>> {
        let handle = self
            .inner
            .as_ref()
            .ok_or_else(|| EggressError::new_err("handle consumed"))?;
        let outcome = py
            .detach(|| handle.reload_toml_str(toml_str))
            .map_err(|e| map_error(py, e))?;
        let dict = PyDict::new(py);
        match outcome {
            eggress_embed::ReloadOutcome::Applied {
                generation,
                upstreams,
            } => {
                dict.set_item("generation", generation)?;
                dict.set_item("upstreams", upstreams)?;
            }
        }
        Ok(dict.into())
    }

    fn shutdown(&mut self, py: Python<'_>) -> PyResult<()> {
        if let Some(handle) = self.inner.take() {
            py.detach(|| handle.shutdown_blocking())
                .map_err(|e| map_error(py, e))?;
        }
        Ok(())
    }

    fn __enter__(slf: Py<Self>) -> Py<Self> {
        slf
    }

    fn __exit__(
        &mut self,
        py: Python<'_>,
        _exc_type: &Bound<'_, PyAny>,
        _exc_value: &Bound<'_, PyAny>,
        _traceback: &Bound<'_, PyAny>,
    ) -> PyResult<bool> {
        if let Some(handle) = self.inner.take() {
            if let Err(e) = py.detach(|| handle.shutdown_blocking()) {
                eprintln!("shutdown error in __exit__: {e}");
            }
        }
        Ok(false)
    }
}

const STATE_CREATED: u8 = 0;
const STATE_CONNECTING: u8 = 1;
const STATE_CONNECTED: u8 = 2;
const STATE_CLOSING: u8 = 3;
const STATE_CLOSED: u8 = 4;
const STATE_FAILED: u8 = 5;

#[pyclass]
struct PyConnection {
    state: Arc<AtomicU8>,
    handle: Option<eggress_embed::EggressHandle>,
    config_toml: String,
    bound_addr: Option<String>,
    remote_addr: Option<String>,
    peername: Option<String>,
    sockname: Option<String>,
    error: Option<String>,
}

#[pymethods]
impl PyConnection {
    #[new]
    #[pyo3(signature = (uris, /, *args))]
    fn new(py: Python<'_>, uris: &Bound<'_, PySequence>, args: Vec<String>) -> PyResult<Self> {
        let mut all_args: Vec<String> = Vec::new();
        for i in 0..uris.len()? {
            all_args.push(uris.get_item(i)?.extract::<String>()?);
        }
        all_args.extend(args);

        if all_args.is_empty() {
            return Err(ConnectionError::new_err(
                "at least one URI argument is required",
            ));
        }

        let parsed = eggress_pproxy_compat::PproxyArgs::parse(&all_args)
            .map_err(|e| ConnectionError::new_err(format!("argument parse error: {e}")))?;

        let output = py
            .detach(|| eggress_pproxy_compat::translate_pproxy_args(&parsed))
            .map_err(|e| ConnectionError::new_err(format!("translation failed: {e}")))?;

        if output.has_unsupported() {
            let features: Vec<_> = output.unsupported.iter().map(|u| u.feature).collect();
            return Err(UnsupportedFeatureError::new_err(format!(
                "unsupported features: {}",
                features.join(", ")
            )));
        }

        let config = eggress_embed::EggressConfig::from_toml_str(&output.toml)
            .map_err(|e| ConnectionError::new_err(format!("config error: {e}")))?;
        let service = eggress_embed::EggressService::new(config);
        let handle = py
            .detach(|| service.start_blocking())
            .map_err(|e| ConnectionError::new_err(format!("startup failed: {e}")))?;

        let addrs = handle.bound_addresses();
        let bound = addrs.listeners.first().map(|l| l.addr.to_string());

        PY_CONNECTION_TOTAL_CREATED.fetch_add(1, Ordering::Relaxed);
        PY_CONNECTION_LIVE_COUNT.fetch_add(1, Ordering::Relaxed);

        Ok(Self {
            state: Arc::new(AtomicU8::new(STATE_CREATED)),
            handle: Some(handle),
            config_toml: output.toml,
            bound_addr: bound,
            remote_addr: None,
            peername: None,
            sockname: None,
            error: None,
        })
    }

    #[getter]
    fn state(&self) -> &str {
        match self.state.load(Ordering::Acquire) {
            STATE_CREATED => "created",
            STATE_CONNECTING => "connecting",
            STATE_CONNECTED => "connected",
            STATE_CLOSING => "closing",
            STATE_CLOSED => "closed",
            STATE_FAILED => "failed",
            _ => "unknown",
        }
    }

    #[getter]
    fn closed(&self) -> bool {
        matches!(
            self.state.load(Ordering::Acquire),
            STATE_CLOSED | STATE_FAILED
        )
    }

    #[getter]
    fn config(&self) -> &str {
        &self.config_toml
    }

    #[getter]
    fn peername(&self) -> Option<&str> {
        self.peername.as_deref()
    }

    #[getter]
    fn sockname(&self) -> Option<&str> {
        self.sockname.as_deref().or(self.bound_addr.as_deref())
    }

    #[getter]
    fn extra_info(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
        let dict = PyDict::new(py);
        dict.set_item("state", self.state())?;
        if let Some(ref addr) = self.bound_addr {
            dict.set_item("bound_addr", addr)?;
        }
        if let Some(ref addr) = self.remote_addr {
            dict.set_item("remote_addr", addr)?;
        }
        if let Some(ref err) = self.error {
            dict.set_item("error", err)?;
        }
        Ok(dict.into())
    }

    fn close(&mut self, py: Python<'_>) -> PyResult<()> {
        if !begin_close(&self.state) {
            return Ok(());
        }
        if let Some(handle) = self.handle.take() {
            py.detach(|| handle.shutdown_blocking())
                .map_err(|e| ConnectionError::new_err(format!("shutdown error: {e}")))?;
        }
        self.state.store(STATE_CLOSED, Ordering::Release);
        let _ = PY_CONNECTION_LIVE_COUNT
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_sub(1));
        Ok(())
    }

    fn wait_closed(&mut self, py: Python<'_>) -> PyResult<()> {
        let current = self.state.load(Ordering::Acquire);
        if current == STATE_CLOSED || current == STATE_FAILED {
            return Ok(());
        }
        self.close(py)?;
        Ok(())
    }

    fn __enter__(slf: Py<Self>) -> Py<Self> {
        slf
    }

    fn __exit__(
        &mut self,
        py: Python<'_>,
        _exc_type: &Bound<'_, PyAny>,
        _exc_value: &Bound<'_, PyAny>,
        _traceback: &Bound<'_, PyAny>,
    ) -> PyResult<bool> {
        self.close(py)?;
        Ok(false)
    }

    fn __del__(&mut self) {
        if !begin_close(&self.state) {
            return;
        }
        eprintln!(
            "Warning: Connection object was not properly closed. Calling close() in __del__."
        );
        if let Some(handle) = self.handle.take() {
            match tokio::runtime::Handle::try_current() {
                Ok(runtime) => {
                    runtime.spawn(async move {
                        if let Err(error) = handle.shutdown().await {
                            eprintln!("shutdown error in __del__: {error}");
                        }
                    });
                }
                Err(error) => {
                    // No runtime available: do not spawn a thread during
                    // interpreter finalization (UB-prone). Drop the handle so
                    // the runtime can reap it; log the deferred shutdown.
                    eprintln!(
                        "could not schedule shutdown in __del__: {error}; \
                         dropping handle to be reaped by runtime"
                    );
                    drop(handle);
                }
            }
        }
        self.state.store(STATE_CLOSED, Ordering::Release);
        // Guard against underflow on double-close paths.
        let _ = PY_CONNECTION_LIVE_COUNT
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| v.checked_sub(1));
    }

    fn __repr__(&self) -> String {
        format!(
            "Connection(state='{}', bound='{}')",
            self.state(),
            self.bound_addr.as_deref().unwrap_or("None")
        )
    }

    #[staticmethod]
    fn connection_stats(py: Python<'_>) -> PyResult<Py<PyDict>> {
        let dict = PyDict::new(py);
        dict.set_item("live", PY_CONNECTION_LIVE_COUNT.load(Ordering::Relaxed))?;
        dict.set_item(
            "total_created",
            PY_CONNECTION_TOTAL_CREATED.load(Ordering::Relaxed),
        )?;
        Ok(dict.into())
    }

    #[staticmethod]
    fn reset_connection_stats() {
        PY_CONNECTION_LIVE_COUNT.store(0, Ordering::Relaxed);
        PY_CONNECTION_TOTAL_CREATED.store(0, Ordering::Relaxed);
    }
}

fn begin_close(state: &AtomicU8) -> bool {
    loop {
        let current = state.load(Ordering::Acquire);
        if current == STATE_CLOSED || current == STATE_FAILED || current == STATE_CLOSING {
            return false;
        }
        match state.compare_exchange(current, STATE_CLOSING, Ordering::AcqRel, Ordering::Acquire) {
            Ok(_) => return true,
            Err(_) => continue,
        }
    }
}

// --- pproxy URI inspection helpers ---

#[pyclass(skip_from_py_object)]
#[derive(Clone)]
struct PyUriInfo {
    scheme: String,
    host: String,
    port: u16,
    tls: bool,
    ssl: bool,
    inbound: bool,
    backward_num: u32,
    has_auth: bool,
    has_rule: bool,
    is_reverse_listener: bool,
    redacted_display: String,
    error: Option<String>,
}

#[pymethods]
impl PyUriInfo {
    #[getter]
    fn scheme(&self) -> &str {
        &self.scheme
    }
    #[getter]
    fn host(&self) -> &str {
        &self.host
    }
    #[getter]
    fn port(&self) -> u16 {
        self.port
    }
    #[getter]
    fn tls(&self) -> bool {
        self.tls
    }
    #[getter]
    fn ssl(&self) -> bool {
        self.ssl
    }
    #[getter]
    fn inbound(&self) -> bool {
        self.inbound
    }
    #[getter]
    fn backward_num(&self) -> u32 {
        self.backward_num
    }
    #[getter]
    fn has_auth(&self) -> bool {
        self.has_auth
    }
    #[getter]
    fn has_rule(&self) -> bool {
        self.has_rule
    }
    #[getter]
    fn is_reverse_listener(&self) -> bool {
        self.is_reverse_listener
    }
    #[getter]
    fn redacted_display(&self) -> &str {
        &self.redacted_display
    }
    #[getter]
    fn error(&self) -> Option<&str> {
        self.error.as_deref()
    }

    fn __repr__(&self) -> String {
        match &self.error {
            Some(e) => format!("UriInfo(error='{}')", e),
            None => format!(
                "UriInfo(scheme='{}', host='{}', port={}, tls={})",
                self.scheme, self.host, self.port, self.tls
            ),
        }
    }
}

#[pyfunction]
fn check_pproxy_uri(uri: &str) -> PyUriInfo {
    match eggress_pproxy_compat::uri::parse_pproxy_uri(uri) {
        Ok(parsed) => PyUriInfo {
            scheme: parsed.scheme.clone(),
            host: parsed.host.clone(),
            port: parsed.port,
            tls: parsed.tls,
            ssl: parsed.ssl,
            inbound: parsed.inbound,
            backward_num: parsed.backward_num,
            has_auth: parsed.username.is_some(),
            has_rule: parsed.rule.is_some(),
            is_reverse_listener: parsed.is_reverse_listener(),
            redacted_display: parsed.redacted_display(),
            error: None,
        },
        Err(e) => PyUriInfo {
            scheme: String::new(),
            host: String::new(),
            port: 0,
            tls: false,
            ssl: false,
            inbound: false,
            backward_num: 0,
            has_auth: false,
            has_rule: false,
            is_reverse_listener: false,
            redacted_display: String::new(),
            error: Some(e.to_string()),
        },
    }
}

#[pyfunction]
fn redact_pproxy_uri(uri: &str) -> PyResult<String> {
    let parsed = eggress_pproxy_compat::uri::parse_pproxy_uri(uri)
        .map_err(|e| UnsupportedFeatureError::new_err(format!("invalid pproxy URI: {e}")))?;
    Ok(parsed.redacted_display())
}

// --- diagnostics ---

#[pyclass(skip_from_py_object)]
#[derive(Clone)]
struct PyDiagnostic {
    code: String,
    feature_id: Option<String>,
    tier: Option<String>,
    message: String,
    suggestion: Option<String>,
}

#[pymethods]
impl PyDiagnostic {
    #[getter]
    fn code(&self) -> &str {
        &self.code
    }
    #[getter]
    fn feature_id(&self) -> Option<&str> {
        self.feature_id.as_deref()
    }
    #[getter]
    fn tier(&self) -> Option<&str> {
        self.tier.as_deref()
    }
    #[getter]
    fn message(&self) -> &str {
        &self.message
    }
    #[getter]
    fn suggestion(&self) -> Option<&str> {
        self.suggestion.as_deref()
    }

    fn __repr__(&self) -> String {
        format!("[{}] {}", self.code, self.message)
    }
}

/// Return diagnostic-only compatibility information for a pproxy URI.
///
/// Entries describe translation warnings and unsupported features; this does
/// not prove that the URI is executable. Use `check_pproxy_uri` for the
/// actionable compatibility check.
#[pyfunction]
fn diagnostics_for_uri(py: Python<'_>, uri: &str) -> PyResult<Vec<PyDiagnostic>> {
    let parsed = eggress_pproxy_compat::uri::parse_pproxy_uri(uri)
        .map_err(|e| UnsupportedFeatureError::new_err(format!("invalid pproxy URI: {e}")))?;

    let mut diagnostics: Vec<PyDiagnostic> = Vec::new();

    let output = py
        .detach(|| {
            eggress_pproxy_compat::translate_from_uris(
                &eggress_pproxy_compat::PproxyArgs::default_args(),
                &[parsed],
                &[],
            )
        })
        .map_err(|e| UnsupportedFeatureError::new_err(format!("translation failed: {e}")))?;

    for warn in &output.warnings {
        let sd = eggress_pproxy_compat::StructuredDiagnostic::from(warn);
        diagnostics.push(PyDiagnostic {
            code: sd.code.to_string(),
            feature_id: sd.feature_id,
            tier: sd.tier,
            message: sd.message,
            suggestion: sd.suggestion,
        });
    }
    for u in &output.unsupported {
        diagnostics.push(PyDiagnostic {
            code: "unsupported_protocol".to_string(),
            feature_id: Some(u.feature.to_string()),
            tier: Some("unsupported".to_string()),
            message: u.detail.clone(),
            suggestion: None,
        });
    }

    diagnostics.sort_by_key(|diagnostic| {
        if diagnostic.tier.as_deref() == Some("unsupported") {
            0
        } else {
            1
        }
    });

    Ok(diagnostics)
}

#[pyfunction]
fn supported_features() -> Vec<&'static str> {
    vec![
        "http",
        "socks4",
        "socks4a",
        "socks5",
        "shadowsocks",
        "trojan",
        "redir",
        "unix",
        "bind",
        "listen",
        "backward",
        "rebind",
        "direct",
    ]
}

// --- config explanation helpers ---

fn parse_toml_config(py: Python<'_>, toml_str: &str) -> PyResult<Py<PyDict>> {
    let parsed: toml::Value = py
        .detach(|| toml::from_str(toml_str))
        .map_err(|e| ConfigError::new_err(format!("failed to parse TOML: {e}")))?;

    let dict = PyDict::new(py);

    // Listeners
    let listeners_list = PyList::empty(py);
    if let Some(listeners) = parsed.get("listeners").and_then(|v| v.as_array()) {
        for l in listeners {
            let ldict = PyDict::new(py);
            if let Some(name) = l.get("name").and_then(|v| v.as_str()) {
                ldict.set_item("name", name)?;
            }
            if let Some(bind) = l.get("bind").and_then(|v| v.as_str()) {
                ldict.set_item("bind", bind)?;
            }
            if let Some(protocols) = l.get("protocols").and_then(|v| v.as_array()) {
                let py_protos = PyList::empty(py);
                for p in protocols {
                    if let Some(s) = p.as_str() {
                        py_protos.append(s)?;
                    }
                }
                ldict.set_item("protocols", py_protos)?;
            }
            if l.get("udp").is_some() {
                ldict.set_item("udp_enabled", true)?;
            }
            if l.get("tls").is_some() {
                ldict.set_item("tls", true)?;
            }
            if l.get("transparent").is_some() {
                ldict.set_item("transparent", true)?;
            }
            if let Some(unix) = l.get("unix") {
                ldict.set_item("unix_socket", true)?;
                if let Some(path) = unix.get("path").and_then(|v| v.as_str()) {
                    ldict.set_item("unix_path", path)?;
                }
            }
            listeners_list.append(ldict)?;
        }
    }
    dict.set_item("listeners", listeners_list)?;

    // Upstreams
    let upstreams_list = PyList::empty(py);
    if let Some(upstreams) = parsed.get("upstreams").and_then(|v| v.as_array()) {
        for u in upstreams {
            let udict = PyDict::new(py);
            if let Some(id) = u.get("id").and_then(|v| v.as_str()) {
                udict.set_item("id", id)?;
            }
            if let Some(uri) = u.get("uri").and_then(|v| v.as_str()) {
                // Redact credentials in the URI
                let redacted = redact_config_uri(uri);
                udict.set_item("uri", redacted)?;
            }
            upstreams_list.append(udict)?;
        }
    }
    dict.set_item("upstreams", upstreams_list)?;

    // Upstream groups
    let groups_list = PyList::empty(py);
    if let Some(groups) = parsed.get("upstream_groups").and_then(|v| v.as_array()) {
        for g in groups {
            let gdict = PyDict::new(py);
            if let Some(id) = g.get("id").and_then(|v| v.as_str()) {
                gdict.set_item("id", id)?;
            }
            if let Some(scheduler) = g.get("scheduler").and_then(|v| v.as_str()) {
                gdict.set_item("scheduler", scheduler)?;
            }
            if let Some(members) = g.get("members").and_then(|v| v.as_array()) {
                let py_members = PyList::empty(py);
                for m in members {
                    if let Some(s) = m.as_str() {
                        py_members.append(s)?;
                    }
                }
                gdict.set_item("members", py_members)?;
            }
            groups_list.append(gdict)?;
        }
    }
    dict.set_item("upstream_groups", groups_list)?;

    // Rules
    let rules_list = PyList::empty(py);
    if let Some(rules) = parsed.get("rules").and_then(|v| v.as_array()) {
        for r in rules {
            let rdict = PyDict::new(py);
            if let Some(id) = r.get("id").and_then(|v| v.as_str()) {
                rdict.set_item("id", id)?;
            }
            if let Some(ug) = r.get("upstream_group").and_then(|v| v.as_str()) {
                rdict.set_item("upstream_group", ug)?;
            }
            if r.get("direct").and_then(|v| v.as_bool()) == Some(true) {
                rdict.set_item("action", "direct")?;
            } else if let Some(reject) = r.get("reject").and_then(|v| v.as_str()) {
                rdict.set_item("action", format!("reject({})", reject))?;
            } else if r.get("upstream_group").is_some() {
                rdict.set_item("action", "upstream")?;
            }
            if r.get("match").is_some() {
                rdict.set_item("has_match", true)?;
            } else if r.get("any").and_then(|v| v.as_bool()) == Some(true) {
                rdict.set_item("match_all", true)?;
            }
            rules_list.append(rdict)?;
        }
    }
    dict.set_item("rules", rules_list)?;

    // Reverse servers
    let reverse_servers_list = PyList::empty(py);
    if let Some(servers) = parsed.get("reverse_servers").and_then(|v| v.as_array()) {
        for s in servers {
            let sdict = PyDict::new(py);
            if let Some(id) = s.get("id").and_then(|v| v.as_str()) {
                sdict.set_item("id", id)?;
            }
            if let Some(bind) = s.get("control_bind").and_then(|v| v.as_str()) {
                sdict.set_item("control_bind", bind)?;
            }
            reverse_servers_list.append(sdict)?;
        }
    }
    dict.set_item("reverse_servers", reverse_servers_list)?;

    // Reverse clients
    let reverse_clients_list = PyList::empty(py);
    if let Some(clients) = parsed.get("reverse_clients").and_then(|v| v.as_array()) {
        for c in clients {
            let cdict = PyDict::new(py);
            if let Some(id) = c.get("id").and_then(|v| v.as_str()) {
                cdict.set_item("id", id)?;
            }
            if let Some(addr) = c.get("server_addr").and_then(|v| v.as_str()) {
                cdict.set_item("server_addr", addr)?;
            }
            reverse_clients_list.append(cdict)?;
        }
    }
    dict.set_item("reverse_clients", reverse_clients_list)?;

    // Security notes
    let security_list = PyList::empty(py);
    // Check for plaintext credentials
    if let Some(listeners) = parsed.get("listeners").and_then(|v| v.as_array()) {
        for l in listeners {
            if l.get("auth").is_some() {
                security_list.append("listener has plaintext auth credentials in TOML")?;
            }
            if l.get("shadowsocks").is_some() {
                security_list.append("listener has Shadowsocks credentials in TOML")?;
            }
        }
    }
    if let Some(servers) = parsed.get("reverse_servers").and_then(|v| v.as_array()) {
        for s in servers {
            if s.get("auth_password").is_some() {
                security_list.append("reverse server has plaintext credentials in TOML")?;
            }
        }
    }
    if let Some(clients) = parsed.get("reverse_clients").and_then(|v| v.as_array()) {
        for c in clients {
            if c.get("auth_password").is_some() {
                security_list.append("reverse client has plaintext credentials in TOML")?;
            }
        }
    }
    if let Some(listeners) = parsed.get("listeners").and_then(|v| v.as_array()) {
        for l in listeners {
            if l.get("transparent").is_some() {
                security_list.append("transparent proxy listener requires elevated privileges")?;
            }
        }
    }
    dict.set_item("security_notes", security_list)?;

    Ok(dict.into())
}

/// Redact credentials from a config URI for safe display.
fn redact_config_uri(uri: &str) -> String {
    eggress_uri::redact_proxy_uri(uri)
}

#[pyfunction]
fn explain_config_toml(py: Python<'_>, toml_str: &str) -> PyResult<Py<PyDict>> {
    parse_toml_config(py, toml_str)
}

#[pyfunction]
fn explain_pproxy_args(py: Python<'_>, args: &Bound<'_, PySequence>) -> PyResult<Py<PyDict>> {
    let result = translate_pproxy_args(py, args)?;
    let toml_str = result.output.toml.clone();
    let warnings: Vec<(String, String)> = result
        .output
        .warnings
        .iter()
        .map(|w| (w.category.to_string(), w.message.clone()))
        .collect();
    let unsupported: Vec<(String, String)> = result
        .output
        .unsupported
        .iter()
        .map(|u| (u.feature.to_string(), u.detail.clone()))
        .collect();
    let is_ok = !result.output.has_unsupported();

    let dict = parse_toml_config(py, &toml_str)?;

    let warnings_list = PyList::empty(py);
    for (cat, msg) in &warnings {
        let wdict = PyDict::new(py);
        wdict.set_item("category", cat.as_str())?;
        wdict.set_item("message", msg.as_str())?;
        warnings_list.append(wdict)?;
    }
    dict.bind(py).set_item("warnings", warnings_list)?;

    let unsupported_list = PyList::empty(py);
    for (feat, detail) in &unsupported {
        let udict = PyDict::new(py);
        udict.set_item("feature", feat.as_str())?;
        udict.set_item("detail", detail.as_str())?;
        unsupported_list.append(udict)?;
    }
    dict.bind(py).set_item("unsupported", unsupported_list)?;

    dict.bind(py).set_item("toml", &toml_str)?;
    dict.bind(py).set_item("ok", is_ok)?;

    Ok(dict)
}

#[pyfunction]
fn explain_pproxy_uri(py: Python<'_>, uri: &str) -> PyResult<Py<PyDict>> {
    let parsed = eggress_pproxy_compat::uri::parse_pproxy_uri(uri)
        .map_err(|e| UnsupportedFeatureError::new_err(format!("invalid pproxy URI: {e}")))?;

    let output = py
        .detach(|| {
            eggress_pproxy_compat::translate_from_uris(
                &eggress_pproxy_compat::PproxyArgs::default_args(),
                &[parsed],
                &[],
            )
        })
        .map_err(|e| UnsupportedFeatureError::new_err(format!("translation failed: {e}")))?;

    let dict = parse_toml_config(py, &output.toml)?;

    let warnings_list = PyList::empty(py);
    for w in &output.warnings {
        let wdict = PyDict::new(py);
        wdict.set_item("category", w.category)?;
        wdict.set_item("message", &w.message)?;
        warnings_list.append(wdict)?;
    }
    dict.bind(py).set_item("warnings", warnings_list)?;

    let unsupported_list = PyList::empty(py);
    for u in &output.unsupported {
        let udict = PyDict::new(py);
        udict.set_item("feature", u.feature)?;
        udict.set_item("detail", &u.detail)?;
        unsupported_list.append(udict)?;
    }
    dict.bind(py).set_item("unsupported", unsupported_list)?;

    dict.bind(py).set_item("toml", &output.toml)?;
    dict.bind(py).set_item("ok", !output.has_unsupported())?;

    Ok(dict)
}

#[pyfunction]
fn route_explain(py: Python<'_>, toml_str: &str, target: &str) -> PyResult<Py<PyDict>> {
    use eggress_config::compile::compile_config;
    use eggress_config::model::ConfigFile;
    use eggress_core::{ClientIdentity, ProtocolId, TargetAddr};
    use eggress_routing::{RouteRequest, Router, TransportKind};

    let target_addr: TargetAddr = target
        .parse()
        .map_err(|e: String| PyValueError::new_err(format!("invalid target: {e}")))?;

    let explanation = py
        .detach(|| -> Result<_, String> {
            let config: ConfigFile =
                toml::from_str(toml_str).map_err(|e| format!("failed to parse TOML: {e}"))?;

            let runtime_config =
                compile_config(&config).map_err(|e| format!("failed to compile config: {e}"))?;

            let router =
                Router::with_groups(runtime_config.rules, runtime_config.default_action, vec![]);

            let request = RouteRequest {
                target: &target_addr,
                source: None,
                listener: "",
                inbound_protocol: ProtocolId::Socks5,
                identity: &ClientIdentity::Anonymous,
                transport: TransportKind::Tcp,
            };

            Ok(router.explain(&request, 0))
        })
        .map_err(ConfigError::new_err)?;

    let dict = PyDict::new(py);
    dict.set_item("target", &explanation.target)?;
    dict.set_item("listener", &explanation.listener)?;
    dict.set_item("protocol", &explanation.protocol)?;
    dict.set_item("transport", &explanation.transport)?;
    dict.set_item("matched_rule", explanation.matched_rule)?;
    dict.set_item("action", &explanation.action)?;
    dict.set_item("upstream_group", explanation.upstream_group)?;
    dict.set_item("scheduler", explanation.scheduler)?;
    let eligible_list = PyList::empty(py);
    for u in &explanation.eligible_upstreams {
        let udict = PyDict::new(py);
        udict.set_item("id", &u.id)?;
        udict.set_item("health", &u.health)?;
        udict.set_item("eligible", u.eligible)?;
        udict.set_item("active", u.active)?;
        udict.set_item("in_flight", u.in_flight)?;
        eligible_list.append(udict)?;
    }
    dict.set_item("eligible_upstreams", eligible_list)?;
    dict.set_item("selected_upstream", explanation.selected_upstream)?;
    dict.set_item("chain", explanation.chain)?;
    dict.set_item("generation", explanation.generation)?;

    Ok(dict.into())
}

#[pyfunction]
fn test_upstream_connect(py: Python<'_>, uri: &str, timeout_secs: f64) -> PyResult<Py<PyDict>> {
    use std::net::ToSocketAddrs;

    let dict = PyDict::new(py);

    // Parse the URI to extract host:port
    let url =
        url::Url::parse(uri).map_err(|e| PyValueError::new_err(format!("invalid URI: {e}")))?;

    let host = url
        .host_str()
        .ok_or_else(|| PyValueError::new_err("URI has no host"))?
        .to_string();
    let port = url.port().unwrap_or(match url.scheme() {
        "socks5" => 1080,
        "socks4" | "socks4a" => 1080,
        "http" | "https" => 80,
        "ss" => 8388,
        "trojan" => 443,
        _ => 0,
    });

    dict.set_item("host", &host)?;
    dict.set_item("port", port)?;
    dict.set_item("scheme", url.scheme())?;

    // Has auth?
    let has_auth = !url.username().is_empty() || url.password().is_some();
    dict.set_item("has_auth", has_auth)?;

    // Redact for display
    let redacted = if has_auth {
        format!("{}://****@{}:{}", url.scheme(), host, port)
    } else {
        format!("{}://{}:{}", url.scheme(), host, port)
    };
    dict.set_item("redacted_uri", &redacted)?;

    // Attempt TCP connect
    let addr_str = format!("{}:{}", host, port);
    let (connected, latency_us, last_error): (bool, Option<u64>, Option<String>) =
        py.detach(|| {
            let std_duration = std::time::Duration::from_secs_f64(timeout_secs);
            let socket_addrs = match addr_str.to_socket_addrs() {
                Ok(addrs) => addrs,
                Err(e) => {
                    return (false, None, Some(format!("DNS resolution failed: {e}")));
                }
            };

            let mut last_error: Option<String> = None;
            for addr in socket_addrs {
                let start = std::time::Instant::now();
                match std::net::TcpStream::connect_timeout(&addr, std_duration) {
                    Ok(_stream) => {
                        return (true, Some(start.elapsed().as_micros() as u64), None);
                    }
                    Err(e) => {
                        last_error = Some(format!("connect to {addr} failed: {e}"));
                    }
                }
            }
            (false, None, last_error)
        });

    dict.set_item("connected", connected)?;
    dict.set_item("latency_us", latency_us)?;
    dict.set_item("error", last_error)?;

    Ok(dict.into())
}

// --- pproxy compatibility translation helpers ---

#[pyclass(skip_from_py_object)]
#[derive(Clone)]
struct PyTranslationWarning {
    inner: eggress_pproxy_compat::CompatWarning,
}

#[pymethods]
impl PyTranslationWarning {
    #[getter]
    fn category(&self) -> &str {
        self.inner.category
    }

    #[getter]
    fn message(&self) -> &str {
        &self.inner.message
    }

    #[getter]
    fn tier(&self) -> &str {
        eggress_pproxy_compat::manifest_tier_for_category(self.inner.category).as_str()
    }

    fn __repr__(&self) -> String {
        format!("[{}] {}", self.inner.category, self.inner.message)
    }
}

#[pyclass(skip_from_py_object)]
#[derive(Clone)]
struct PyUnsupportedFeature {
    inner: eggress_pproxy_compat::UnsupportedFeature,
}

#[pymethods]
impl PyUnsupportedFeature {
    #[getter]
    fn feature(&self) -> &str {
        self.inner.feature
    }

    #[getter]
    fn message(&self) -> &str {
        &self.inner.detail
    }

    #[getter]
    fn tier(&self) -> &str {
        eggress_pproxy_compat::classify_unsupported_feature_tier(self.inner.feature)
    }

    fn __repr__(&self) -> String {
        format!("unsupported {}: {}", self.inner.feature, self.inner.detail)
    }
}

#[pyclass]
struct PyTranslationResult {
    output: eggress_pproxy_compat::TranslationOutput,
}

#[pymethods]
impl PyTranslationResult {
    #[getter]
    fn toml(&self) -> &str {
        &self.output.toml
    }

    #[getter]
    fn warnings(&self, py: Python<'_>) -> PyResult<Py<PyList>> {
        let list = PyList::empty(py);
        for w in &self.output.warnings {
            list.append(PyTranslationWarning { inner: w.clone() })?;
        }
        Ok(list.into())
    }

    #[getter]
    fn unsupported(&self, py: Python<'_>) -> PyResult<Py<PyList>> {
        let list = PyList::empty(py);
        for u in &self.output.unsupported {
            list.append(PyUnsupportedFeature { inner: u.clone() })?;
        }
        Ok(list.into())
    }

    #[getter]
    fn ok(&self) -> bool {
        !self.output.has_unsupported()
    }

    /// Manifest-aligned aggregate compatibility tier.
    ///
    /// This is the single source of truth for the five-tier compatibility
    /// classification; Python and the canonical manifest must agree with
    /// this value. The same value is reused by the CLI `pproxy check`
    /// reporter.
    #[getter]
    fn tier(&self) -> &'static str {
        eggress_pproxy_compat::classify_aggregate_tier(
            &self.output.warnings,
            &self.output.unsupported,
        )
        .as_str()
    }

    fn config(&self, py: Python<'_>) -> PyResult<PyEggressConfig> {
        let config = py
            .detach(|| eggress_embed::EggressConfig::from_toml_str(&self.output.toml))
            .map_err(|e| map_error(py, e))?;
        Ok(PyEggressConfig { inner: config })
    }

    fn __repr__(&self) -> String {
        format!(
            "TranslationResult(warnings={}, unsupported={})",
            self.output.warnings.len(),
            self.output.unsupported.len()
        )
    }
}

#[pyfunction]
fn translate_pproxy_args(
    py: Python<'_>,
    args: &Bound<'_, PySequence>,
) -> PyResult<PyTranslationResult> {
    let len = args.len()?;
    let raw: Vec<String> = (0..len)
        .map(|i| args.get_item(i)?.extract::<String>())
        .collect::<PyResult<_>>()?;

    let parsed = eggress_pproxy_compat::PproxyArgs::parse(&raw).map_err(|e| {
        UnsupportedFeatureError::new_err(format!("failed to parse pproxy args: {e}"))
    })?;

    let output = py
        .detach(|| eggress_pproxy_compat::translate_pproxy_args(&parsed))
        .map_err(|e| UnsupportedFeatureError::new_err(format!("translation failed: {e}")))?;

    Ok(PyTranslationResult { output })
}

#[pyfunction]
fn translate_pproxy_uri(
    py: Python<'_>,
    local: &str,
    remotes: Option<&Bound<'_, PySequence>>,
) -> PyResult<PyTranslationResult> {
    let local_uri = eggress_pproxy_compat::uri::parse_pproxy_uri(local)
        .map_err(|e| UnsupportedFeatureError::new_err(format!("invalid local URI: {e}")))?;

    let remote_chains: Vec<eggress_pproxy_compat::PproxyChain> = match remotes {
        Some(seq) => {
            let len = seq.len()?;
            (0..len)
                .map(|i| {
                    let s: String = seq.get_item(i)?.extract()?;
                    eggress_pproxy_compat::uri::parse_pproxy_chain(&s).map_err(|e| {
                        UnsupportedFeatureError::new_err(format!("invalid remote URI: {e}"))
                    })
                })
                .collect::<PyResult<_>>()?
        }
        None => Vec::new(),
    };

    let output = py
        .detach(|| {
            eggress_pproxy_compat::translate_from_uris(
                &eggress_pproxy_compat::PproxyArgs::default_args(),
                &[local_uri],
                &remote_chains,
            )
        })
        .map_err(|e| UnsupportedFeatureError::new_err(format!("translation failed: {e}")))?;

    Ok(PyTranslationResult { output })
}

#[pyfunction]
fn check_pproxy_args(
    py: Python<'_>,
    args: &Bound<'_, PySequence>,
) -> PyResult<PyTranslationResult> {
    translate_pproxy_args(py, args)
}

/// Validate the frozen executable parser contract without starting a service.
/// Migration helpers intentionally retain their broader translation-only
/// extension surface.
#[pyfunction]
fn validate_pproxy_args(args: &Bound<'_, PySequence>) -> PyResult<()> {
    let len = args.len()?;
    let raw: Vec<String> = (0..len)
        .map(|i| args.get_item(i)?.extract::<String>())
        .collect::<PyResult<_>>()?;
    let parsed = if raw.is_empty() {
        eggress_pproxy_compat::PproxyArgs::default_args()
    } else {
        eggress_pproxy_compat::PproxyArgs::parse(&raw)
            .map_err(|e| PyValueError::new_err(format!("pproxy argument error: {e}")))?
    };
    if let Some(flag) = parsed.strict_parser_violations().first() {
        return Err(PyValueError::new_err(format!(
            "pproxy: unknown option or positional argument '{flag}'"
        )));
    }
    parsed
        .validate_strict_values()
        .map_err(|e| PyValueError::new_err(format!("pproxy argument error: {e}")))
}

/// Return compatibility runtime options using the canonical Rust parser.
/// Migration callers may still use the broader translation surface; strict
/// executable entry points perform their separate parser-gate validation.
#[pyfunction]
fn pproxy_runtime_options(args: &Bound<'_, PySequence>) -> PyResult<Py<PyDict>> {
    let len = args.len()?;
    let raw: Vec<String> = (0..len)
        .map(|i| args.get_item(i)?.extract::<String>())
        .collect::<PyResult<_>>()?;
    let parsed = if raw.is_empty() {
        eggress_pproxy_compat::PproxyArgs::default_args()
    } else {
        eggress_pproxy_compat::PproxyArgs::parse(&raw)
            .map_err(|e| PyValueError::new_err(format!("pproxy argument error: {e}")))?
    };
    let dict = PyDict::new(args.py());
    dict.set_item(
        "auth_timeout_seconds",
        parsed.effective_auth_timeout().as_secs(),
    )?;
    dict.set_item("system_proxy", parsed.system_proxy)?;
    dict.set_item("debug", parsed.debug)?;
    dict.set_item("verbose_level", parsed.verbose_level)?;
    dict.set_item("default_log_level", parsed.default_log_level())?;
    Ok(dict.into())
}

/// Install compatibility-process logging for `python -m pproxy` while
/// respecting an existing embedded application's tracing subscriber.
#[pyfunction]
fn init_pproxy_logging(default_level: &str) -> PyResult<()> {
    let filter = tracing_subscriber::EnvFilter::try_from_default_env()
        .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(default_level));
    let _ = tracing_subscriber::fmt()
        .with_env_filter(filter)
        .compact()
        .try_init();
    Ok(())
}

/// Run the native compatibility upstream test without starting listeners.
#[pyfunction]
fn run_pproxy_test(py: Python<'_>, args: &Bound<'_, PySequence>, target: &str) -> PyResult<i32> {
    let len = args.len()?;
    let raw: Vec<String> = (0..len)
        .map(|i| args.get_item(i)?.extract::<String>())
        .collect::<PyResult<_>>()?;
    let parsed = if raw.is_empty() {
        eggress_pproxy_compat::PproxyArgs::default_args()
    } else {
        eggress_pproxy_compat::PproxyArgs::parse(&raw)
            .map_err(|e| PyValueError::new_err(format!("pproxy argument error: {e}")))?
    };
    if let Some(flag) = parsed.strict_parser_violations().first() {
        return Err(PyValueError::new_err(format!(
            "pproxy: unknown option or positional argument '{flag}'"
        )));
    }
    parsed
        .validate_strict_values()
        .map_err(|e| PyValueError::new_err(format!("pproxy argument error: {e}")))?;
    let output = eggress_pproxy_compat::translate_pproxy_args(&parsed)
        .map_err(|e| PyValueError::new_err(format!("pproxy translation error: {e}")))?;
    let gate = eggress_pproxy_compat::evaluate_execution_gate(&parsed, &output);
    if !gate.allows_start() {
        return Err(UnsupportedFeatureError::new_err(gate.blocker_summary()));
    }
    let (config, _) = eggress_config::validate_and_compile_toml_with_warnings(&output.toml)
        .map_err(|e| ConfigError::new_err(format!("pproxy config error: {e}")))?;
    let target = eggress_cli::parse_pproxy_test_target(target)
        .map_err(PyValueError::new_err)?
        .to_string();
    if config.upstreams.is_empty() {
        return Ok(0);
    }
    Ok(py.detach(|| {
        eggress_cli::run_upstream_test(&config, Some(&target), Duration::from_secs(10), false)
    }))
}

#[pyclass]
struct PyReverseUriSummary {
    /// "server" or "client" or "unknown"
    role: String,
    scheme: String,
    /// "host:port" string in redacted form for display
    target: String,
    has_auth: bool,
    /// "reverse_servers" or "reverse_clients" or "unknown"
    toml_section: String,
    tls: bool,
    /// Modifiers parsed (e.g. "+tls", "+in")
    modifiers: Vec<String>,
}

#[pymethods]
impl PyReverseUriSummary {
    #[getter]
    fn role(&self) -> &str {
        &self.role
    }
    #[getter]
    fn scheme(&self) -> &str {
        &self.scheme
    }
    #[getter]
    fn target(&self) -> &str {
        &self.target
    }
    #[getter]
    fn has_auth(&self) -> bool {
        self.has_auth
    }
    #[getter]
    fn toml_section(&self) -> &str {
        &self.toml_section
    }
    #[getter]
    fn tls(&self) -> bool {
        self.tls
    }
    #[getter]
    fn modifiers(&self) -> Vec<String> {
        self.modifiers.clone()
    }
    fn __repr__(&self) -> String {
        format!(
            "ReverseUriSummary(role={}, target={}, toml_section={}, has_auth={})",
            self.role, self.target, self.toml_section, self.has_auth
        )
    }
}

#[pyfunction]
fn describe_reverse_pproxy_uri(uri: &str) -> PyResult<PyReverseUriSummary> {
    let parsed = eggress_pproxy_compat::uri::parse_pproxy_uri(uri)
        .map_err(|e| UnsupportedFeatureError::new_err(format!("invalid pproxy URI: {e}")))?;

    let (role, toml_section) = if parsed.is_reverse_listener() {
        ("server", "reverse_servers")
    } else if parsed.is_backward() {
        ("client", "reverse_clients")
    } else {
        ("unknown", "unknown")
    };

    let target = parsed.redacted_display();

    // Modifiers encoded in the scheme: +tls, +ssl, +in, ...
    let mut modifiers: Vec<String> = Vec::new();
    if parsed.tls {
        modifiers.push("+tls".to_string());
    }
    if parsed.ssl {
        modifiers.push("+ssl".to_string());
    }
    for _ in 0..parsed.backward_num {
        modifiers.push("+in".to_string());
    }

    Ok(PyReverseUriSummary {
        role: role.to_string(),
        scheme: parsed.scheme.clone(),
        target,
        has_auth: parsed.username.is_some(),
        toml_section: toml_section.to_string(),
        tls: parsed.tls,
        modifiers,
    })
}

/// Python-accessible outbound connector for direct proxy chain connections.
///
/// This wraps the Rust `OutboundConnector` and provides a Python interface
/// for making outbound TCP connections through a configured proxy chain
/// without starting a listener service.
///
/// The returned stream owns the Tokio runtime needed by the Rust transport,
/// so it remains usable after the connector is dropped. Blocking methods
/// release the Python GIL while they wait for network I/O.
#[pyclass]
struct PyOutboundConnector {
    inner: eggress_embed::outbound::OutboundConnector,
}

struct OutboundStreamState {
    stream: Option<eggress_core::BoxStream>,
}

/// A connected native outbound stream.
///
/// This deliberately exposes a small socket/asyncio-stream compatible surface
/// instead of a file descriptor. Advanced transports (TLS, WebSocket, H2,
/// and multi-hop chains) do not necessarily have a meaningful OS socket after
/// the first hop. `recv`/`sendall` aliases are provided for pproxy programs;
/// `read`/`write` are the canonical APIs.
#[pyclass]
struct PyOutboundStream {
    runtime: Arc<tokio::runtime::Runtime>,
    state: std::sync::Mutex<OutboundStreamState>,
    peer_addr: Option<String>,
    local_addr: Option<String>,
    hop_count: usize,
}

impl PyOutboundStream {
    fn with_stream<T>(
        &self,
        operation: impl FnOnce(&mut eggress_core::BoxStream) -> T,
    ) -> PyResult<T> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| ConnectionError::new_err("outbound stream lock poisoned"))?;
        let stream = state
            .stream
            .as_mut()
            .ok_or_else(|| ConnectionClosedError::new_err("outbound stream is closed"))?;
        Ok(operation(stream))
    }

    fn closed_inner(&self) -> bool {
        self.state
            .lock()
            .map(|state| state.stream.is_none())
            .unwrap_or(true)
    }
}

impl Drop for PyOutboundStream {
    fn drop(&mut self) {
        // Dropping the BoxStream closes the transport. Do not block from a
        // destructor: Python may be shutting down and the runtime can be
        // unavailable at that point. Use try_lock to avoid blocking the
        // Python GC while another thread holds the state mutex.
        if let Ok(mut state) = self.state.try_lock() {
            state.stream.take();
        }
    }
}

#[pymethods]
impl PyOutboundStream {
    #[getter]
    fn closed(&self) -> bool {
        self.closed_inner()
    }

    fn is_closing(&self) -> bool {
        self.closed_inner()
    }

    #[getter]
    fn peername(&self) -> Option<&str> {
        self.peer_addr.as_deref()
    }

    #[getter]
    fn sockname(&self) -> Option<&str> {
        self.local_addr.as_deref()
    }

    fn get_extra_info(
        &self,
        py: Python<'_>,
        name: &str,
        default: Option<Py<PyAny>>,
    ) -> PyResult<Py<PyAny>> {
        match name {
            "peername" => match self.peer_addr.as_ref() {
                Some(value) => Ok(PyString::new(py, value).into_any().unbind()),
                None => Ok(default.unwrap_or_else(|| py.None())),
            },
            "sockname" => match self.local_addr.as_ref() {
                Some(value) => Ok(PyString::new(py, value).into_any().unbind()),
                None => Ok(default.unwrap_or_else(|| py.None())),
            },
            "hop_count" => {
                let hop_count = self.hop_count.to_string();
                Ok(PyString::new(py, &hop_count).into_any().unbind())
            }
            _ => Ok(default.unwrap_or_else(|| py.None())),
        }
    }

    const MAX_READ_LEN: usize = 16 * 1024 * 1024;

    fn read(&self, py: Python<'_>, n: i64) -> PyResult<Vec<u8>> {
        if n < -1 {
            return Err(PyValueError::new_err(
                "read length must be -1 or non-negative",
            ));
        }
        let len = if n == -1 {
            None
        } else {
            let len = usize::try_from(n).map_err(|_| {
                PyValueError::new_err(format!(
                    "read length does not fit in usize on this platform (n={n})"
                ))
            })?;
            if len > Self::MAX_READ_LEN {
                return Err(PyValueError::new_err(format!(
                    "read length {len} exceeds maximum {}",
                    Self::MAX_READ_LEN
                )));
            }
            Some(len)
        };
        let runtime = self.runtime.clone();
        self.with_stream(|stream| {
            py.detach(|| {
                runtime.block_on(async {
                    if let Some(len) = len {
                        let mut data = vec![0_u8; len];
                        let count = stream.read(&mut data).await?;
                        data.truncate(count);
                        Ok(data)
                    } else {
                        let mut data = Vec::new();
                        let mut limited = stream.take(Self::MAX_READ_LEN as u64);
                        limited.read_to_end(&mut data).await.map(|_| data)
                    }
                })
            })
        })?
        .map_err(|e: std::io::Error| ConnectionError::new_err(format!("read failed: {e}")))
    }

    fn readexactly(&self, py: Python<'_>, n: usize) -> PyResult<Vec<u8>> {
        if n > Self::MAX_READ_LEN {
            return Err(PyValueError::new_err(format!(
                "readexactly length {n} exceeds maximum {}",
                Self::MAX_READ_LEN
            )));
        }
        let runtime = self.runtime.clone();
        self.with_stream(|stream| {
            py.detach(|| {
                runtime.block_on(async {
                    let mut data = vec![0_u8; n];
                    stream.read_exact(&mut data).await.map(|_| data)
                })
            })
        })?
        .map_err(|e: std::io::Error| ConnectionError::new_err(format!("readexactly failed: {e}")))
    }

    fn write(&self, py: Python<'_>, data: &[u8]) -> PyResult<usize> {
        let runtime = self.runtime.clone();
        self.with_stream(|stream| {
            py.detach(|| runtime.block_on(async { stream.write(data).await }))
        })?
        .map_err(|e: std::io::Error| ConnectionError::new_err(format!("write failed: {e}")))
    }

    fn sendall(&self, py: Python<'_>, data: &[u8]) -> PyResult<()> {
        let runtime = self.runtime.clone();
        self.with_stream(|stream| {
            py.detach(|| runtime.block_on(async { stream.write_all(data).await }))
        })?
        .map_err(|e: std::io::Error| ConnectionError::new_err(format!("sendall failed: {e}")))
    }

    fn drain(&self, py: Python<'_>) -> PyResult<()> {
        let runtime = self.runtime.clone();
        self.with_stream(|stream| py.detach(|| runtime.block_on(async { stream.flush().await })))?
            .map_err(|e: std::io::Error| ConnectionError::new_err(format!("drain failed: {e}")))
    }

    fn write_eof(&self, py: Python<'_>) -> PyResult<()> {
        let runtime = self.runtime.clone();
        self.with_stream(|stream| {
            py.detach(|| runtime.block_on(async { stream.shutdown().await }))
        })?
        .map_err(|e: std::io::Error| ConnectionError::new_err(format!("write_eof failed: {e}")))
    }

    fn close(&self) -> PyResult<()> {
        let mut state = self
            .state
            .lock()
            .map_err(|_| ConnectionError::new_err("outbound stream lock poisoned"))?;
        state.stream.take();
        Ok(())
    }

    fn wait_closed(&self) -> PyResult<()> {
        self.close()
    }

    fn __enter__(slf: Py<Self>) -> Py<Self> {
        slf
    }

    fn __exit__(
        &self,
        _py: Python<'_>,
        _exc_type: &Bound<'_, PyAny>,
        _exc_value: &Bound<'_, PyAny>,
        _traceback: &Bound<'_, PyAny>,
    ) -> PyResult<bool> {
        self.close()?;
        Ok(false)
    }

    fn __del__(&self) {
        if !self.closed_inner() {
            eprintln!("Warning: outbound stream was not explicitly closed; cleaning up");
            let _ = self.close();
        }
    }

    fn __repr__(&self) -> String {
        format!(
            "OutboundStream(peername={:?}, hop_count={}, closed={})",
            self.peer_addr,
            self.hop_count,
            self.closed_inner()
        )
    }
}

#[pymethods]
impl PyOutboundConnector {
    /// Create a connector from a pproxy-style URI string.
    #[staticmethod]
    fn from_pproxy_uri(uri: &str) -> PyResult<Self> {
        let inner = eggress_embed::outbound::OutboundConnector::from_pproxy_uri(uri)
            .map_err(|e| ConnectionError::new_err(format!("failed to create connector: {e}")))?;
        Ok(Self { inner })
    }

    /// Create a connector from a TOML config string.
    #[staticmethod]
    fn from_toml(config_toml: &str) -> PyResult<Self> {
        let inner = eggress_embed::outbound::OutboundConnector::from_toml(config_toml)
            .map_err(|e| ConnectionError::new_err(format!("failed to create connector: {e}")))?;
        Ok(Self { inner })
    }

    /// Validate that a TOML config is usable for outbound connections.
    ///
    /// Returns the number of hops in the first upstream's chain.
    #[staticmethod]
    fn validate_config(config_toml: &str) -> PyResult<usize> {
        eggress_embed::outbound::OutboundConnector::validate_outbound_config(config_toml)
            .map_err(|e| ConnectionError::new_err(format!("config validation failed: {e}")))
    }

    /// Get the number of upstreams configured.
    fn upstream_count(&self) -> usize {
        self.inner.upstream_count()
    }

    /// Open a native outbound TCP stream. No local listener is created.
    #[pyo3(signature = (host, port, timeout=None))]
    fn connect_tcp(
        &self,
        py: Python<'_>,
        host: &str,
        port: u16,
        timeout: Option<f64>,
    ) -> PyResult<PyOutboundStream> {
        if host.is_empty() {
            return Err(PyValueError::new_err("host must not be empty"));
        }
        let timeout = timeout
            .map(|seconds| {
                if !seconds.is_finite() || seconds <= 0.0 {
                    Err(PyValueError::new_err("timeout must be finite and positive"))
                } else {
                    Ok(Duration::from_secs_f64(seconds))
                }
            })
            .transpose()?;
        let runtime = outbound_runtime().map_err(ConnectionError::new_err)?;
        let inner = &self.inner;
        let result = py.detach(|| {
            runtime.block_on(async {
                match timeout {
                    Some(timeout) => inner.connect_tcp_timeout(host, port, timeout).await,
                    None => inner.connect_tcp(host, port).await,
                }
            })
        });
        let (stream, info) = result
            .map_err(|e| ConnectionError::new_err(format!("outbound connect failed: {e}")))?;
        Ok(PyOutboundStream {
            runtime,
            state: std::sync::Mutex::new(OutboundStreamState {
                stream: Some(stream),
            }),
            peer_addr: info.peer_addr.map(|addr| addr.to_string()),
            local_addr: info.local_addr.map(|addr| addr.to_string()),
            hop_count: info.hop_count,
        })
    }

    /// Get connection metadata for a target host:port.
    ///
    /// Resolves the first hop endpoint and returns metadata about the
    /// configured chain without actually connecting.
    fn preview_connect(&self, py: Python<'_>, host: &str, port: u16) -> PyResult<Py<PyDict>> {
        let dict = PyDict::new(py);
        dict.set_item("target_host", host)?;
        dict.set_item("target_port", port)?;
        dict.set_item("hop_count", self.inner.upstream_count())?;
        Ok(dict.into())
    }
}

#[pymodule]
fn _eggress(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_class::<PyEggressConfig>()?;
    m.add_class::<PyEggressService>()?;
    m.add_class::<PyEggressHandle>()?;
    m.add_class::<PyAppliedSystemProxy>()?;
    m.add_class::<PyTranslationWarning>()?;
    m.add_class::<PyUnsupportedFeature>()?;
    m.add_class::<PyTranslationResult>()?;
    m.add_class::<PyReverseUriSummary>()?;
    m.add_class::<PyUriInfo>()?;
    m.add_class::<PyDiagnostic>()?;
    m.add_class::<PyConnection>()?;
    m.add_class::<PyOutboundConnector>()?;
    m.add_class::<PyOutboundStream>()?;
    m.add_function(wrap_pyfunction!(translate_pproxy_args, m)?)?;
    m.add_function(wrap_pyfunction!(translate_pproxy_uri, m)?)?;
    m.add_function(wrap_pyfunction!(check_pproxy_args, m)?)?;
    m.add_function(wrap_pyfunction!(validate_pproxy_args, m)?)?;
    m.add_function(wrap_pyfunction!(pproxy_runtime_options, m)?)?;
    m.add_function(wrap_pyfunction!(init_pproxy_logging, m)?)?;
    m.add_function(wrap_pyfunction!(run_pproxy_test, m)?)?;
    m.add_function(wrap_pyfunction!(describe_reverse_pproxy_uri, m)?)?;
    m.add_function(wrap_pyfunction!(check_pproxy_uri, m)?)?;
    m.add_function(wrap_pyfunction!(redact_pproxy_uri, m)?)?;
    m.add_function(wrap_pyfunction!(diagnostics_for_uri, m)?)?;
    m.add_function(wrap_pyfunction!(supported_features, m)?)?;
    m.add_function(wrap_pyfunction!(explain_config_toml, m)?)?;
    m.add_function(wrap_pyfunction!(explain_pproxy_args, m)?)?;
    m.add_function(wrap_pyfunction!(explain_pproxy_uri, m)?)?;
    m.add_function(wrap_pyfunction!(route_explain, m)?)?;
    m.add_function(wrap_pyfunction!(test_upstream_connect, m)?)?;
    m.add_function(wrap_pyfunction!(apply_system_proxy, m)?)?;
    m.add("EggressError", m.py().get_type::<EggressError>())?;
    m.add("ConfigError", m.py().get_type::<ConfigError>())?;
    m.add("StartupError", m.py().get_type::<StartupError>())?;
    m.add("ReloadError", m.py().get_type::<ReloadError>())?;
    m.add("ShutdownError", m.py().get_type::<ShutdownError>())?;
    m.add(
        "UnsupportedFeatureError",
        m.py().get_type::<UnsupportedFeatureError>(),
    )?;
    m.add("InternalError", m.py().get_type::<InternalError>())?;
    m.add("ConnectionError", m.py().get_type::<ConnectionError>())?;
    m.add(
        "ConnectionClosedError",
        m.py().get_type::<ConnectionClosedError>(),
    )?;
    m.add("TimeoutError", m.py().get_type::<TimeoutError>())?;
    m.add("DnsError", m.py().get_type::<DnsError>())?;
    m.add("AuthError", m.py().get_type::<AuthError>())?;
    m.add("TlsError", m.py().get_type::<TlsError>())?;
    m.add("LoopMismatchError", m.py().get_type::<LoopMismatchError>())?;
    m.add(
        "ConnectionCancelledError",
        m.py().get_type::<ConnectionCancelledError>(),
    )?;
    m.add(
        "UseAfterCloseError",
        m.py().get_type::<UseAfterCloseError>(),
    )?;
    m.add(
        "UdpAssociationError",
        m.py().get_type::<UdpAssociationError>(),
    )?;
    m.add(
        "UnsupportedCompositionError",
        m.py().get_type::<UnsupportedCompositionError>(),
    )?;
    m.add("__version__", env!("CARGO_PKG_VERSION"))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    #[test]
    fn uri_translation_smoke() {
        let parsed = eggress_pproxy_compat::uri::parse_pproxy_uri("http://127.0.0.1:8080")
            .expect("valid pproxy URI");
        assert_eq!(parsed.scheme, "http");
        assert_eq!(parsed.port, 8080);

        let args = eggress_pproxy_compat::PproxyArgs::parse(&["http://127.0.0.1:8080".to_string()])
            .expect("valid pproxy args");
        let output =
            eggress_pproxy_compat::translate_pproxy_args(&args).expect("translatable pproxy args");
        assert!(output.toml.contains("127.0.0.1:8080"));
    }

    #[test]
    fn config_conversion_smoke() {
        let args = eggress_pproxy_compat::PproxyArgs::parse(&["http://127.0.0.1:8080".to_string()])
            .expect("valid pproxy args");
        let output =
            eggress_pproxy_compat::translate_pproxy_args(&args).expect("translatable pproxy args");
        let config = eggress_embed::EggressConfig::from_toml_str(&output.toml)
            .expect("translated TOML converts to embed config");
        assert!(!config.source_toml().is_empty());
    }

    #[test]
    fn error_mapping_categories_are_stable() {
        use eggress_embed::EggressError;

        let cases = [
            (EggressError::Config("bad config".into()), "config"),
            (EggressError::Runtime("runtime".into()), "runtime"),
            (EggressError::Startup("startup".into()), "startup"),
            (EggressError::Reload("reload".into()), "reload"),
            (EggressError::Shutdown("shutdown".into()), "shutdown"),
            (
                EggressError::UnsupportedFeature {
                    feature: "feature".into(),
                    message: "unsupported".into(),
                },
                "unsupported_feature",
            ),
            (EggressError::Internal("internal".into()), "internal"),
        ];
        for (error, category) in cases {
            assert_eq!(error.category(), category);
        }
    }
}