rmux 0.4.3

A local terminal multiplexer with a tmux-style CLI, daemon runtime, Rust SDK, and ratatui integration.
#![cfg(windows)]

use std::error::Error;
use std::ffi::OsString;
use std::path::Path;
use std::process::{Child, Command, ExitStatus, Output, Stdio};
use std::sync::{Mutex, OnceLock};
use std::thread;
use std::time::{Duration, Instant};

use rmux_client::{
    connect, ensure_server_running, socket_path_for_label, Connection, INTERNAL_DAEMON_FLAG,
};
use rmux_proto::{KillServerRequest, ListSessionsRequest, Request, Response};

const BINARY_OVERRIDE_ENV: &str = "RMUX_INTERNAL_BINARY_PATH";
const BINARY_OVERRIDE_TEST_OPT_IN_ENV: &str = "RMUX_ALLOW_INTERNAL_BINARY_OVERRIDE";
const CLIENT_VERSION_OVERRIDE_ENV: &str = "RMUX_INTERNAL_CLIENT_VERSION";

#[test]
fn hidden_daemon_mode_serves_windows_ipc_requests() -> Result<(), Box<dyn Error>> {
    let socket_path =
        socket_path_for_label(format!("hidden-daemon-windows-{}", std::process::id()))?;
    let mut child = Command::new(env!("CARGO_BIN_EXE_rmux"))
        .arg(INTERNAL_DAEMON_FLAG)
        .arg(&socket_path)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()?;

    let mut connection = match wait_for_connection(&socket_path, &mut child) {
        Ok(connection) => connection,
        Err(error) => {
            let _ = child.kill();
            let _ = child.wait();
            return Err(error);
        }
    };

    let response = connection.roundtrip(&Request::ListSessions(ListSessionsRequest {
        format: None,
        filter: None,
        sort_order: None,
        reversed: false,
    }))?;
    let Response::ListSessions(response) = response else {
        let _ = child.kill();
        let _ = child.wait();
        return Err("list-sessions did not return a list-sessions response".into());
    };
    assert!(response.output.stdout().is_empty());

    let response = connection.roundtrip(&Request::KillServer(KillServerRequest))?;
    assert!(matches!(response, Response::KillServer(_)));
    wait_for_child_exit(&mut child)?;
    Ok(())
}

#[test]
fn ensure_server_running_auto_starts_windows_hidden_daemon() -> Result<(), Box<dyn Error>> {
    let _guard = env_lock().lock().expect("lock env");
    let previous_binary = std::env::var_os(BINARY_OVERRIDE_ENV);
    let previous_opt_in = std::env::var_os(BINARY_OVERRIDE_TEST_OPT_IN_ENV);
    let _binary_restore = EnvVarGuard::new(BINARY_OVERRIDE_ENV, previous_binary);
    let _opt_in_restore = EnvVarGuard::new(BINARY_OVERRIDE_TEST_OPT_IN_ENV, previous_opt_in);
    std::env::set_var(BINARY_OVERRIDE_ENV, env!("CARGO_BIN_EXE_rmux"));
    std::env::set_var(BINARY_OVERRIDE_TEST_OPT_IN_ENV, "1");
    let socket_path = socket_path_for_label(format!("auto-start-windows-{}", std::process::id()))?;

    let mut connection = ensure_server_running(&socket_path)?;
    let response = connection.roundtrip(&Request::ListSessions(ListSessionsRequest {
        format: None,
        filter: None,
        sort_order: None,
        reversed: false,
    }))?;
    assert!(matches!(response, Response::ListSessions(_)));

    let response = connection.roundtrip(&Request::KillServer(KillServerRequest))?;
    assert!(matches!(response, Response::KillServer(_)));
    Ok(())
}

#[test]
fn start_server_with_captured_output_returns_after_spawning_windows_daemon(
) -> Result<(), Box<dyn Error>> {
    let _guard = env_lock().lock().expect("lock env");
    let previous_binary = std::env::var_os(BINARY_OVERRIDE_ENV);
    let previous_opt_in = std::env::var_os(BINARY_OVERRIDE_TEST_OPT_IN_ENV);
    let _binary_restore = EnvVarGuard::new(BINARY_OVERRIDE_ENV, previous_binary);
    let _opt_in_restore = EnvVarGuard::new(BINARY_OVERRIDE_TEST_OPT_IN_ENV, previous_opt_in);
    std::env::set_var(BINARY_OVERRIDE_ENV, env!("CARGO_BIN_EXE_rmux"));
    std::env::set_var(BINARY_OVERRIDE_TEST_OPT_IN_ENV, "1");

    let socket_path = socket_path_for_label(format!(
        "captured-start-server-windows-{}",
        std::process::id()
    ))?;
    let output = run_rmux_command(&socket_path, &["start-server"])?;
    assert!(
        output.status.success(),
        "captured start-server failed: status={:?}\nstdout={}\nstderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(output.stdout.is_empty());
    assert!(output.stderr.is_empty());

    let output = run_rmux_command(&socket_path, &["kill-server"])?;
    assert!(
        output.status.success(),
        "captured kill-server failed: status={:?}\nstdout={}\nstderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    wait_for_daemon_process_absent(&socket_path)?;
    Ok(())
}

#[test]
fn seamless_upgrade_restarts_idle_stale_windows_daemon() -> Result<(), Box<dyn Error>> {
    let _guard = env_lock().lock().expect("lock env");
    let socket_path =
        socket_path_for_label(format!("seamless-upgrade-windows-{}", std::process::id()))?;
    let mut old_daemon = spawn_hidden_daemon(&socket_path)?;
    let old_connection = match wait_for_connection(&socket_path, &mut old_daemon) {
        Ok(connection) => connection,
        Err(error) => {
            let _ = old_daemon.kill();
            let _ = old_daemon.wait();
            return Err(error);
        }
    };
    drop(old_connection);

    let status = run_client_as_newer_version(&socket_path, &["start-server"])?;
    assert!(
        status.success(),
        "newer client failed against idle stale daemon: status={:?}",
        status.code()
    );

    wait_for_child_exit(&mut old_daemon)?;
    let output = run_rmux_command(&socket_path, &["list-sessions"])?;
    assert!(
        output.status.success(),
        "new daemon did not serve detached RPC after seamless upgrade: status={:?}\nstdout={}\nstderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    let output = run_rmux_command(&socket_path, &["kill-server"])?;
    assert!(
        output.status.success(),
        "new daemon did not accept kill-server after seamless upgrade: status={:?}\nstdout={}\nstderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    wait_for_daemon_process_absent(&socket_path)?;
    Ok(())
}

#[test]
fn seamless_upgrade_preserves_active_stale_windows_daemon() -> Result<(), Box<dyn Error>> {
    let _guard = env_lock().lock().expect("lock env");
    let socket_path =
        socket_path_for_label(format!("seamless-active-windows-{}", std::process::id()))?;
    let mut old_daemon = spawn_hidden_daemon(&socket_path)?;
    let mut active_connection = match wait_for_connection(&socket_path, &mut old_daemon) {
        Ok(connection) => connection,
        Err(error) => {
            let _ = old_daemon.kill();
            let _ = old_daemon.wait();
            return Err(error);
        }
    };

    let status = run_client_as_newer_version(&socket_path, &["start-server"])?;
    assert!(
        status.success(),
        "newer client failed against active stale daemon: status={:?}",
        status.code()
    );
    assert!(
        old_daemon.try_wait()?.is_none(),
        "active stale daemon must not be replaced while it owns another client connection"
    );

    let response = active_connection.roundtrip(&Request::KillServer(KillServerRequest))?;
    assert!(matches!(response, Response::KillServer(_)));
    drop(active_connection);
    wait_for_child_exit(&mut old_daemon)?;
    Ok(())
}

fn spawn_hidden_daemon(socket_path: &Path) -> Result<Child, Box<dyn Error>> {
    Ok(Command::new(env!("CARGO_BIN_EXE_rmux"))
        .arg(INTERNAL_DAEMON_FLAG)
        .arg(socket_path)
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()?)
}

fn run_client_as_newer_version(
    socket_path: &Path,
    args: &[&str],
) -> Result<ExitStatus, Box<dyn Error>> {
    let mut command = Command::new(env!("CARGO_BIN_EXE_rmux"));
    command
        .arg("-S")
        .arg(socket_path)
        .args(args)
        .env(BINARY_OVERRIDE_ENV, env!("CARGO_BIN_EXE_rmux"))
        .env(BINARY_OVERRIDE_TEST_OPT_IN_ENV, "1")
        .env(CLIENT_VERSION_OVERRIDE_ENV, "999.0.0-test");
    run_command_status(command)
}

fn run_rmux_command(socket_path: &Path, args: &[&str]) -> Result<Output, Box<dyn Error>> {
    let mut command = Command::new(env!("CARGO_BIN_EXE_rmux"));
    command.arg("-S").arg(socket_path).args(args);
    run_command_output(command)
}

fn run_command_status(mut command: Command) -> Result<ExitStatus, Box<dyn Error>> {
    let mut child = command
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()?;
    let deadline = Instant::now() + Duration::from_secs(10);

    loop {
        if let Some(status) = child.try_wait()? {
            return Ok(status);
        }
        if Instant::now() >= deadline {
            let _ = child.kill();
            let status = child.wait()?;
            return Err(format!("rmux command timed out: status={:?}", status.code()).into());
        }
        thread::sleep(Duration::from_millis(25));
    }
}

fn run_command_output(mut command: Command) -> Result<Output, Box<dyn Error>> {
    let mut child = command
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    let deadline = Instant::now() + Duration::from_secs(10);

    loop {
        if child.try_wait()?.is_some() {
            return Ok(child.wait_with_output()?);
        }
        if Instant::now() >= deadline {
            let _ = child.kill();
            let output = child.wait_with_output()?;
            return Err(format!(
                "rmux command timed out: status={:?}\nstdout={}\nstderr={}",
                output.status.code(),
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            )
            .into());
        }
        thread::sleep(Duration::from_millis(25));
    }
}

fn wait_for_connection(
    socket_path: &std::path::Path,
    child: &mut Child,
) -> Result<Connection, Box<dyn Error>> {
    let deadline = Instant::now() + Duration::from_secs(5);

    loop {
        if let Some(status) = child.try_wait()? {
            return Err(format!("hidden daemon exited before accepting IPC: {status}").into());
        }

        match connect(socket_path) {
            Ok(connection) => return Ok(connection),
            Err(error) if Instant::now() < deadline => {
                let _ = error;
                thread::sleep(Duration::from_millis(25));
            }
            Err(error) => return Err(Box::new(error)),
        }
    }
}

fn wait_for_daemon_process_absent(socket_path: &Path) -> Result<(), Box<dyn Error>> {
    let deadline = Instant::now() + Duration::from_secs(5);

    loop {
        let process_count = daemon_process_count(socket_path)?;
        if process_count == 0 {
            return Ok(());
        }
        if Instant::now() >= deadline {
            return Err(format!(
                "{process_count} hidden daemon process(es) still match the test pipe after kill-server"
            )
            .into());
        }
        thread::sleep(Duration::from_millis(25));
    }
}

fn daemon_process_count(socket_path: &Path) -> Result<usize, Box<dyn Error>> {
    let script = r#"
$needle = $env:RMUX_TEST_PIPE
$count = @(
  Get-CimInstance Win32_Process |
    Where-Object { $_.Name -eq 'rmux.exe' -and $_.CommandLine -like "*$needle*" }
).Count
Write-Output $count
"#;
    let output = Command::new("powershell")
        .args([
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-Command",
            script,
        ])
        .env("RMUX_TEST_PIPE", socket_path)
        .stdin(Stdio::null())
        .output()?;
    assert!(
        output.status.success(),
        "failed to query daemon processes: status={:?}\nstderr={}",
        output.status.code(),
        String::from_utf8_lossy(&output.stderr)
    );
    let stdout = String::from_utf8(output.stdout)?;
    Ok(stdout.trim().parse()?)
}

fn wait_for_child_exit(child: &mut Child) -> Result<(), Box<dyn Error>> {
    let deadline = Instant::now() + Duration::from_secs(5);

    loop {
        if child.try_wait()?.is_some() {
            return Ok(());
        }
        if Instant::now() >= deadline {
            child.kill()?;
            let _ = child.wait();
            return Err("hidden daemon did not exit after kill-server".into());
        }
        thread::sleep(Duration::from_millis(25));
    }
}

fn env_lock() -> &'static Mutex<()> {
    static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    LOCK.get_or_init(|| Mutex::new(()))
}

struct EnvVarGuard {
    name: &'static str,
    previous_value: Option<OsString>,
}

impl EnvVarGuard {
    fn new(name: &'static str, previous_value: Option<OsString>) -> Self {
        Self {
            name,
            previous_value,
        }
    }
}

impl Drop for EnvVarGuard {
    fn drop(&mut self) {
        match self.previous_value.as_ref() {
            Some(value) => std::env::set_var(self.name, value),
            None => std::env::remove_var(self.name),
        }
    }
}