openhuman/app/src-tauri/src/core_process.rs

use std::path::PathBuf;
use std::sync::Arc;

use tokio::net::TcpStream;
use tokio::process::{Child, Command};
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tokio::time::{timeout, Duration};

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CoreRunMode {
    InProcess,
    ChildProcess,
}

#[derive(Clone)]
pub struct CoreProcessHandle {
    child: Arc<Mutex<Option<Child>>>,
    task: Arc<Mutex<Option<JoinHandle<()>>>>,
    restart_lock: Arc<Mutex<()>>,
    port: u16,
    core_bin: Option<PathBuf>,
    run_mode: CoreRunMode,
}

impl CoreProcessHandle {
    pub fn new(port: u16, core_bin: Option<PathBuf>, run_mode: CoreRunMode) -> Self {
        Self {
            child: Arc::new(Mutex::new(None)),
            task: Arc::new(Mutex::new(None)),
            restart_lock: Arc::new(Mutex::new(())),
            port,
            core_bin,
            run_mode,
        }
    }

    pub fn rpc_url(&self) -> String {
        format!("http://127.0.0.1:{}/rpc", self.port)
    }

    /// Acquire the restart lock to serialize overlapping restart requests.
    pub async fn restart_lock(&self) -> tokio::sync::MutexGuard<'_, ()> {
        self.restart_lock.lock().await
    }

    async fn is_rpc_port_open(&self) -> bool {
        matches!(
            timeout(
                Duration::from_millis(150),
                TcpStream::connect(("127.0.0.1", self.port)),
            )
            .await,
            Ok(Ok(_))
        )
    }

    pub async fn ensure_running(&self) -> Result<(), String> {
        if self.is_rpc_port_open().await {
            log::info!(
                "[core] found existing core rpc endpoint at {}",
                self.rpc_url()
            );
            return Ok(());
        }

        match self.run_mode {
            CoreRunMode::InProcess => {
                log::warn!(
                    "[core] in-process core mode is unavailable in host-only build; falling back to child process"
                );
                let mut guard = self.child.lock().await;
                if guard.is_none() {
                    let mut cmd = if let Some(core_bin) = &self.core_bin {
                        let mut cmd = Command::new(core_bin);
                        if is_current_exe_path(core_bin) {
                            cmd.arg("core");
                        }
                        cmd.arg("run").arg("--port").arg(self.port.to_string());
                        cmd
                    } else {
                        let exe = std::env::current_exe()
                            .map_err(|e| format!("failed to resolve current executable: {e}"))?;
                        let mut cmd = Command::new(exe);
                        cmd.arg("core")
                            .arg("run")
                            .arg("--port")
                            .arg(self.port.to_string());
                        cmd
                    };
                    let child = cmd
                        .spawn()
                        .map_err(|e| format!("failed to spawn core process: {e}"))?;
                    *guard = Some(child);
                }
            }
            CoreRunMode::ChildProcess => {
                let mut guard = self.child.lock().await;
                if guard.is_none() {
                    let mut cmd = if let Some(core_bin) = &self.core_bin {
                        let mut cmd = Command::new(core_bin);
                        if is_current_exe_path(core_bin) {
                            // Safety: if core_bin resolves to this GUI executable, force the
                            // explicit subcommand path so we don't accidentally relaunch clients.
                            cmd.arg("core");
                        }
                        cmd.arg("run").arg("--port").arg(self.port.to_string());
                        log::info!(
                            "[core] spawning dedicated core binary: {:?} run --port {}",
                            cmd.as_std().get_program(),
                            self.port
                        );
                        cmd
                    } else {
                        let exe = std::env::current_exe()
                            .map_err(|e| format!("failed to resolve current executable: {e}"))?;
                        let mut cmd = Command::new(exe);
                        cmd.arg("core")
                            .arg("run")
                            .arg("--port")
                            .arg(self.port.to_string());
                        log::warn!(
                            "[core] dedicated core binary not found; falling back to self subcommand"
                        );
                        cmd
                    };

                    let child = cmd
                        .spawn()
                        .map_err(|e| format!("failed to spawn core process: {e}"))?;

                    *guard = Some(child);
                }
            }
        }

        for _ in 0..40 {
            if self.is_rpc_port_open().await {
                log::info!("[core] core rpc became ready at {}", self.rpc_url());
                return Ok(());
            }

            match self.run_mode {
                CoreRunMode::InProcess => {
                    let mut guard = self.task.lock().await;
                    if let Some(task) = guard.as_ref() {
                        if task.is_finished() {
                            let task = guard.take().expect("checked is_some");
                            drop(guard);
                            match task.await {
                                Ok(_) => {
                                    return Err(
                                        "in-process core server exited before becoming ready"
                                            .to_string(),
                                    )
                                }
                                Err(err) => {
                                    return Err(format!(
                                        "in-process core server task failed before ready: {err}"
                                    ))
                                }
                            }
                        }
                    }
                }
                CoreRunMode::ChildProcess => {
                    let mut guard = self.child.lock().await;
                    if let Some(child) = guard.as_mut() {
                        match child.try_wait() {
                            Ok(Some(status)) => {
                                return Err(format!("core process exited before ready: {status}"));
                            }
                            Ok(None) => {}
                            Err(e) => {
                                return Err(format!("failed checking core process status: {e}"));
                            }
                        }
                    }
                }
            }
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
        }

        Err("core process did not become ready".to_string())
    }

    /// Restart the core process to pick up updated macOS permission grants.
    ///
    /// macOS caches permission state per-process; the running sidecar never sees
    /// a newly granted permission until it restarts. This method shuts down the
    /// current child, waits until the RPC port is free (so `ensure_running` does not
    /// fast-return while the old listener is still bound), then spawns a fresh instance.
    ///
    /// If another process is listening on the core port (e.g. manual `openhuman core run`),
    /// shutdown does not stop it — we time out and return an error instead of a false success.
    ///
    /// Issue: <https://github.com/tinyhumansai/openhuman/issues/133>
    pub async fn restart(&self) -> Result<(), String> {
        log::info!("[core] restarting core process for permission refresh");

        let had_managed_child = {
            let guard = self.child.lock().await;
            guard.is_some()
        };
        log::debug!(
            "[core] restart: had_managed_child={} before shutdown",
            had_managed_child
        );

        self.shutdown().await;
        log::debug!("[core] restart: shutdown complete, checking port {}", self.port);

        // If we never spawned the sidecar (something else was already listening), we cannot free
        // the port — fail fast with a clear message instead of polling for 8s.
        if !had_managed_child && self.is_rpc_port_open().await {
            log::error!(
                "[core] restart: no child to stop but port {} is open — another process owns it",
                self.port
            );
            return Err(format!(
                "Core RPC port {} is already in use by another process (OpenHuman did not start it). Quit any `openhuman core run` in a terminal or free the port, then relaunch the app. You can also set OPENHUMAN_CORE_PORT to a different port.",
                self.port
            ));
        }

        // After kill+wait on our child, the port should close; poll briefly in case the OS is slow
        // to release the socket.
        const POLL_MS: u64 = 50;
        const MAX_WAIT_MS: u64 = 10_000;
        let mut waited_ms: u64 = 0;
        while self.is_rpc_port_open().await {
            if waited_ms >= MAX_WAIT_MS {
                log::error!(
                    "[core] restart: port {} still in use after {}ms (had_managed_child={})",
                    self.port,
                    MAX_WAIT_MS,
                    had_managed_child
                );
                return Err(format!(
                    "Core RPC port {} did not become free after stopping the sidecar. Quit any other process using this port (e.g. `openhuman core run`) or change OPENHUMAN_CORE_PORT.",
                    self.port
                ));
            }
            tokio::time::sleep(std::time::Duration::from_millis(POLL_MS)).await;
            waited_ms += POLL_MS;
        }

        log::debug!("[core] restart: port free, calling ensure_running");
        let result = self.ensure_running().await;
        match &result {
            Ok(()) => log::info!("[core] restart: core process ready after restart"),
            Err(e) => log::error!("[core] restart: failed to restart core process: {e}"),
        }
        result
    }

    /// Stop the core process this handle spawned (child or in-process task). Safe to call if
    /// nothing was spawned or core was already external.
    pub async fn shutdown(&self) {
        let mut child_guard = self.child.lock().await;
        if let Some(mut child) = child_guard.take() {
            log::info!("[core] terminating child core process on app shutdown");
            if let Err(e) = child.kill().await {
                log::warn!("[core] failed to kill child core process: {e}");
            }
            // Wait for the process to exit so the RPC listen socket is released before restart
            // checks the port (otherwise we can spuriously hit "port still in use").
            match timeout(
                Duration::from_secs(12),
                child.wait(),
            )
            .await
            {
                Ok(Ok(status)) => {
                    log::debug!("[core] child core process reaped after kill: {status}");
                }
                Ok(Err(e)) => {
                    log::warn!("[core] wait on child core process after kill: {e}");
                }
                Err(_) => {
                    log::warn!(
                        "[core] timed out waiting for child core process to exit after kill (12s)"
                    );
                }
            }
        }
        let mut task_guard = self.task.lock().await;
        if let Some(task) = task_guard.take() {
            task.abort();
        }
    }
}

fn is_current_exe_path(candidate: &std::path::Path) -> bool {
    let Ok(current) = std::env::current_exe() else {
        return false;
    };
    same_executable_path(candidate, &current)
}

fn same_executable_path(a: &std::path::Path, b: &std::path::Path) -> bool {
    if a == b {
        return true;
    }
    match (std::fs::canonicalize(a), std::fs::canonicalize(b)) {
        (Ok(a_real), Ok(b_real)) => a_real == b_real,
        _ => false,
    }
}

pub fn default_core_port() -> u16 {
    std::env::var("OPENHUMAN_CORE_PORT")
        .ok()
        .and_then(|v| v.parse::<u16>().ok())
        .unwrap_or(7788)
}

pub fn default_core_run_mode(_daemon_mode: bool) -> CoreRunMode {
    if let Ok(value) = std::env::var("OPENHUMAN_CORE_RUN_MODE") {
        let normalized = value.trim().to_ascii_lowercase();
        if matches!(normalized.as_str(), "inprocess" | "in-process" | "internal") {
            return CoreRunMode::InProcess;
        }
        if matches!(
            normalized.as_str(),
            "child" | "process" | "external" | "sidecar"
        ) {
            return CoreRunMode::ChildProcess;
        }
    }

    // Default to a dedicated core process so app and core lifecycles are separated.
    CoreRunMode::ChildProcess
}

pub fn default_core_bin() -> Option<PathBuf> {
    if let Ok(path) = std::env::var("OPENHUMAN_CORE_BIN") {
        let candidate = PathBuf::from(path);
        if candidate.exists() {
            return Some(candidate);
        }
    }

    // Dev: prefer a staged sidecar under src-tauri/binaries, then use the same search as
    // release (next to the .app, Resources/, etc.). Previously we returned None here when the
    // folder was empty, which forced `core run` on the GUI binary — a different TCC identity than
    // `openhuman-core-*` and misleading "still denied" after granting the sidecar name.
    #[cfg(debug_assertions)]
    {
        let binaries_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("binaries");
        if let Ok(entries) = std::fs::read_dir(&binaries_dir) {
            for entry in entries.flatten() {
                let path = entry.path();
                if !path.is_file() {
                    continue;
                }
                let Some(file_name) = path.file_name().and_then(|n| n.to_str()) else {
                    continue;
                };
                #[cfg(windows)]
                let matches =
                    file_name.starts_with("openhuman-core-") && file_name.ends_with(".exe");
                #[cfg(not(windows))]
                let matches = file_name.starts_with("openhuman-core-");
                if matches {
                    return Some(path);
                }
            }
        }
    }

    let exe = std::env::current_exe().ok()?;
    let exe_dir = exe.parent()?;

    #[cfg(windows)]
    let standalone = exe_dir.join("openhuman-core.exe");
    #[cfg(not(windows))]
    let standalone = exe_dir.join("openhuman-core");

    if standalone.exists() && !same_executable_path(&standalone, &exe) {
        return Some(standalone);
    }

    #[cfg(windows)]
    let legacy_standalone = exe_dir.join("openhuman-core.exe");
    #[cfg(not(windows))]
    let legacy_standalone = exe_dir.join("openhuman-core");

    if legacy_standalone.exists() && !same_executable_path(&legacy_standalone, &exe) {
        return Some(legacy_standalone);
    }

    // Sidecar layout: bundle.externalBin("binaries/openhuman-core") is emitted as
    // openhuman-core-<target-triple>(.exe) under app resources.
    let search_dirs = {
        let mut dirs = vec![exe_dir.to_path_buf()];
        #[cfg(target_os = "macos")]
        {
            if let Some(resources_dir) = exe_dir.parent().map(|p| p.join("Resources")) {
                dirs.push(resources_dir);
            }
        }
        dirs
    };

    for dir in search_dirs {
        let Ok(entries) = std::fs::read_dir(&dir) else {
            continue;
        };
        for entry in entries.flatten() {
            let path = entry.path();
            if !path.is_file() {
                continue;
            }
            let Some(file_name) = path.file_name().and_then(|n| n.to_str()) else {
                continue;
            };

            #[cfg(windows)]
            let matches = (file_name.starts_with("openhuman-core-") && file_name.ends_with(".exe"))
                || (file_name.starts_with("openhuman-core-") && file_name.ends_with(".exe"));
            #[cfg(not(windows))]
            let matches = file_name.starts_with("openhuman-core-")
                || file_name.starts_with("openhuman-core-");

            if matches && !same_executable_path(&path, &exe) {
                return Some(path);
            }
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use super::{
        default_core_port, default_core_run_mode, same_executable_path, CoreProcessHandle,
        CoreRunMode,
    };

    struct EnvGuard {
        key: &'static str,
        old: Option<String>,
    }

    impl EnvGuard {
        fn set(key: &'static str, value: &str) -> Self {
            let old = std::env::var(key).ok();
            std::env::set_var(key, value);
            Self { key, old }
        }

        fn unset(key: &'static str) -> Self {
            let old = std::env::var(key).ok();
            std::env::remove_var(key);
            Self { key, old }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            if let Some(old) = &self.old {
                std::env::set_var(self.key, old);
            } else {
                std::env::remove_var(self.key);
            }
        }
    }

    #[test]
    fn default_core_run_mode_env_parsing() {
        let _guard = EnvGuard::set("OPENHUMAN_CORE_RUN_MODE", "in-process");
        assert_eq!(default_core_run_mode(false), CoreRunMode::InProcess);

        let _guard = EnvGuard::set("OPENHUMAN_CORE_RUN_MODE", "sidecar");
        assert_eq!(default_core_run_mode(false), CoreRunMode::ChildProcess);
    }

    #[test]
    fn default_core_port_env_and_fallback() {
        let _unset = EnvGuard::unset("OPENHUMAN_CORE_PORT");
        assert_eq!(default_core_port(), 7788);

        let _set = EnvGuard::set("OPENHUMAN_CORE_PORT", "8899");
        assert_eq!(default_core_port(), 8899);
    }

    #[test]
    fn same_executable_path_handles_equal_and_non_equal_paths() {
        let current = std::env::current_exe().expect("current exe");
        assert!(same_executable_path(&current, &current));

        let different = current.with_file_name("definitely-not-the-current-exe");
        assert!(!same_executable_path(&current, &different));
    }

    #[test]
    fn ensure_running_returns_ok_when_rpc_port_already_open() {
        let rt = tokio::runtime::Runtime::new().expect("runtime");
        let result = rt.block_on(async {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
                .await
                .expect("bind test listener");
            let port = listener.local_addr().expect("local addr").port();
            let handle = CoreProcessHandle::new(port, None, CoreRunMode::ChildProcess);
            handle.ensure_running().await
        });
        assert!(
            result.is_ok(),
            "ensure_running should fast-path: {result:?}"
        );
    }
}