cua-sdk 0.9.0

//! Native (non-wasm) implementation.

mod agent_setup;
mod agents;
mod auth;
mod bake;
#[cfg(feature = "spaces")]
mod cloud;
mod config;
mod cua;
#[cfg(feature = "spaces")]
mod elastic;
#[cfg(feature = "host")]
mod host;
mod image;
#[cfg(feature = "local")]
mod local;
mod mcp;
#[cfg(feature = "media")]
mod media;
mod overlay;
#[cfg(feature = "spaces")]
mod persistent;
mod sandbox;
mod share;
#[cfg(feature = "spaces")]
mod spaces;
mod spacesd;
pub(crate) mod telemetry;
#[cfg(feature = "spaces")]
mod teleport_types;

pub use agent_setup::*;
pub use agents::*;
pub use auth::*;
pub use bake::*;
#[cfg(feature = "spaces")]
pub use cloud::*;
pub use config::*;
pub use cua::*;
#[cfg(feature = "spaces")]
pub use elastic::*;
#[cfg(feature = "host")]
pub use host::*;
pub use image::*;
#[cfg(feature = "local")]
pub use local::*;
pub use mcp::*;
#[cfg(feature = "media")]
pub use media::*;
pub use overlay::*;
#[cfg(feature = "spaces")]
pub use persistent::*;
pub use sandbox::*;
pub use share::*;
pub use spaces::*;
pub use spacesd::*;
pub use telemetry::*;
#[cfg(feature = "spaces")]
pub use teleport_types::*;

use crate::CuaError;
use std::{future::Future, sync::OnceLock};

/// The Tokio runtime every exported async call runs on. Foreign executors
/// only poll the join handle, so callers never need a runtime of their own
/// and background tasks (forwards, media pumps) outlive the call.
pub(crate) fn runtime() -> &'static tokio::runtime::Runtime {
    static RT: OnceLock<tokio::runtime::Runtime> = OnceLock::new();
    RT.get_or_init(|| {
        // A trainer holds a connection (or more) per sandbox: at macOS's
        // default soft limit of 256 its opens fail past ~250 sandboxes
        // ("Too many open files").
        let _ = raise_open_file_limit();
        let workers = std::thread::available_parallelism()
            .map(|n| n.get().clamp(2, 4))
            .unwrap_or(2);
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(workers)
            .thread_name("cua-sdk")
            .enable_all()
            .build()
            .expect("build the cua-sdk Tokio runtime")
    })
}

/// Runs `fut` on the SDK runtime and awaits it from any executor.
pub(crate) async fn run<T, F>(fut: F) -> crate::Result<T>
where
    T: Send + 'static,
    F: Future<Output = crate::Result<T>> + Send + 'static,
{
    runtime()
        .spawn(fut)
        .await
        .map_err(|e| CuaError::Internal(format!("SDK task failed: {e}")))?
}

/// A not-found message as [`CuaError::NotFound`] carries it: its display
/// says "not found: …" already, so a cua-spaces message that ends with "not
/// found" ("{what} not found") loses that ending instead of saying it twice.
pub(crate) fn not_found_message(message: String) -> String {
    match message.strip_suffix(" not found") {
        Some(what) if !what.trim().is_empty() => what.to_string(),
        _ => message,
    }
}

impl From<cua_daemon::Error> for CuaError {
    fn from(e: cua_daemon::Error) -> Self {
        use cua_daemon::Error as E;
        let m = e.message().to_string();
        match e {
            E::InvalidArgument(_) => CuaError::InvalidArgument(m),
            E::InvalidPlacement { .. } => CuaError::InvalidPlacement(m),
            E::NotFound(_) => CuaError::NotFound(not_found_message(m)),
            E::AmbiguousSandbox { .. } => CuaError::AmbiguousSandbox(m),
            E::ProviderNotConfigured(_) => CuaError::ProviderNotConfigured(m),
            E::Unsupported(_) => CuaError::Unsupported(m),
            E::SpacesdNotAvailable(_) => CuaError::SpacesdNotAvailable(m),
            E::Timeout(_) => CuaError::Timeout(m),
            E::Cloud(c) => CuaError::Cloud(c),
            E::Runtime(_) => CuaError::Runtime(m),
            E::Env(_) => CuaError::Env(m),
            E::Http(_) => CuaError::Http(m),
            E::Unauthenticated(_) => CuaError::Unauthenticated(m),
            E::Transport(_) => CuaError::Transport(m),
            E::DaemonNotRunning(_) => CuaError::DaemonNotRunning(m),
            E::CapabilityMissing(_) => CuaError::CapabilityMissing(m),
            E::HostCapabilityMissing(_) => CuaError::HostCapabilityMissing(m),
            E::TeleportRefused(_) => CuaError::TeleportRefused(m),
            E::InsufficientDisk(_) => CuaError::InsufficientDisk(m),
            E::HostCapacity(_) => CuaError::HostCapacity(m),
            E::Cancelled(_) => CuaError::Cancelled(m),
            E::Internal(_) => CuaError::Internal(m),
        }
    }
}

impl From<cua_spacesd_client::Error> for CuaError {
    fn from(e: cua_spacesd_client::Error) -> Self {
        use cua_spacesd_client::Error as E;
        let m = e.to_string();
        match e {
            E::InvalidEndpoint(_) => CuaError::InvalidArgument(m),
            E::Transport(_) => CuaError::Transport(m),
            E::SpacesdNotAvailable { .. } => CuaError::SpacesdNotAvailable(m),
            E::Unauthenticated(_) => CuaError::Unauthenticated(m),
            E::FeatureUnsupported { .. } => CuaError::Unsupported(m),
            E::PermissionDenied(_) => CuaError::PermissionDenied(m),
            E::ProcessNotFound(_) | E::PathNotFound(_) | E::SessionNotFound(_) => {
                CuaError::NotFound(m)
            }
            E::Timeout(_) | E::DesktopNotReady(_) => CuaError::Timeout(m),
            E::Io(_) => CuaError::Internal(m),
            _ => CuaError::Env(m),
        }
    }
}

impl From<cua_sandbox_core::Error> for CuaError {
    fn from(e: cua_sandbox_core::Error) -> Self {
        cua_daemon::Error::from(e).into()
    }
}

impl From<std::io::Error> for CuaError {
    fn from(e: std::io::Error) -> Self {
        CuaError::Internal(e.to_string())
    }
}

impl From<tonic::Status> for CuaError {
    fn from(s: tonic::Status) -> Self {
        cua_spacesd_client::Error::from(s).into()
    }
}

pub(crate) fn ms(v: Option<u32>, default_ms: u32) -> std::time::Duration {
    std::time::Duration::from_millis(u64::from(v.unwrap_or(default_ms)))
}

pub(crate) fn millis(v: impl Into<u64>) -> std::time::Duration {
    std::time::Duration::from_millis(v.into())
}

pub(crate) fn secs(v: impl Into<u64>) -> std::time::Duration {
    std::time::Duration::from_secs(v.into())
}

/// Test helper: tests that change the process environment take turns.
#[cfg(test)]
pub(crate) mod test_env {
    /// A private `CUA_HOME` and no `CUA_DEFAULT_*` for one test; tests that
    /// touch the process environment take turns.
    pub(crate) struct EnvGuard {
        _lock: std::sync::MutexGuard<'static, ()>,
        home: tempfile::TempDir,
        saved: Vec<(&'static str, Option<String>)>,
    }

    static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
    const VARS: [&str; 4] = [
        "CUA_HOME",
        "CUA_DEFAULT_ON",
        "CUA_DEFAULT_KIND",
        "CUA_DEFAULT_RUNTIME",
    ];

    impl EnvGuard {
        pub(crate) fn isolated() -> Self {
            let lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
            let home = tempfile::tempdir().unwrap();
            let saved = VARS.iter().map(|v| (*v, std::env::var(v).ok())).collect();
            // SAFETY: tests that change the environment hold ENV_LOCK.
            unsafe {
                for v in VARS {
                    std::env::remove_var(v);
                }
                std::env::set_var("CUA_HOME", home.path());
            }
            Self {
                _lock: lock,
                home,
                saved,
            }
        }

        pub(crate) fn set(&self, k: &str, v: &str) {
            let _ = &self.home;
            // SAFETY: see `isolated`.
            unsafe { std::env::set_var(k, v) }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            // SAFETY: see `isolated`.
            unsafe {
                for (k, v) in &self.saved {
                    match v {
                        Some(v) => std::env::set_var(k, v),
                        None => std::env::remove_var(k),
                    }
                }
            }
        }
    }
}

#[cfg(all(test, feature = "spaces"))]
mod not_found_tests {
    use super::*;

    /// "Run on" a device that is no host: the create's error says "not
    /// found" once, whether it came from the embedded runtime or the daemon.
    #[test]
    fn a_missing_machine_says_not_found_once() {
        let what = "a computer matching \"dev_22\" among your computers (\"gamma-4 Mac Studio\" (host:96fe))";
        let embedded: CuaError = cua_spaces::Error::NotFound(what.into()).into();
        assert_eq!(embedded.to_string(), format!("not found: {what}"));
        let daemon: CuaError = cua_daemon::Error::NotFound(format!("{what} not found")).into();
        assert_eq!(daemon.to_string(), format!("not found: {what}"));
        // A message that is only the word stays as it is.
        assert_eq!(not_found_message(" not found".into()), " not found");
        assert_eq!(not_found_message("local:box".into()), "local:box");
    }
}

/// The open-file limit an SDK process (and the daemon) asks for: it holds a
/// socket or pipe per Space, client, stream and engine connection, and
/// macOS starts processes (launchd, a terminal) at a soft limit of 256.
#[cfg(unix)]
const WANTED_OPEN_FILES: u64 = 65_536;

/// Raises this process's soft open-file limit toward 65,536, within the
/// hard limit (and, on macOS, `kern.maxfilesperproc`). Returns the soft
/// limit before and after, or `None` when nothing changed.
#[cfg(unix)]
pub fn raise_open_file_limit() -> Option<(u64, u64)> {
    let mut lim = libc::rlimit {
        rlim_cur: 0,
        rlim_max: 0,
    };
    // SAFETY: getrlimit fills the struct it is given.
    if unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) } != 0 {
        return None;
    }
    let before: u64 = lim.rlim_cur;
    #[allow(unused_mut)]
    let mut ceiling = lim.rlim_max.min(WANTED_OPEN_FILES);
    #[cfg(target_os = "macos")]
    {
        // setrlimit refuses more than this on macOS, whatever the hard limit.
        let mut per_proc: libc::c_int = 0;
        let mut len = std::mem::size_of::<libc::c_int>();
        // SAFETY: sysctlbyname writes at most `len` bytes into `per_proc`.
        let rc = unsafe {
            libc::sysctlbyname(
                c"kern.maxfilesperproc".as_ptr(),
                (&mut per_proc as *mut libc::c_int).cast(),
                &mut len,
                std::ptr::null_mut(),
                0,
            )
        };
        if rc == 0 && per_proc > 0 {
            ceiling = ceiling.min(per_proc as u64);
        }
    }
    if ceiling <= before {
        return None;
    }
    lim.rlim_cur = ceiling;
    // SAFETY: setrlimit reads the struct it is given.
    if unsafe { libc::setrlimit(libc::RLIMIT_NOFILE, &lim) } != 0 {
        return None;
    }
    Some((before, ceiling))
}

/// Not applicable off Unix.
#[cfg(not(unix))]
pub fn raise_open_file_limit() -> Option<(u64, u64)> {
    None
}

#[cfg(all(test, unix))]
mod fd_limit_tests {
    #[test]
    fn the_sdk_runtime_raises_the_open_file_limit() {
        let _ = super::runtime();
        let mut lim = libc::rlimit {
            rlim_cur: 0,
            rlim_max: 0,
        };
        // SAFETY: getrlimit fills the struct it is given.
        assert_eq!(unsafe { libc::getrlimit(libc::RLIMIT_NOFILE, &mut lim) }, 0);
        assert!(lim.rlim_cur >= lim.rlim_max.min(1024), "{}", lim.rlim_cur);
        // Raising again is a no-op.
        assert_eq!(super::raise_open_file_limit(), None);
    }
}