rmux 0.7.1

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

use std::error::Error;
use std::io;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};

use rmux_pty::{
    write_windows_console_key, ChildCommand, SpawnedPty, TerminalSize, WindowsConsoleKeyEvent,
};

const SETUP_TIMEOUT: Duration = Duration::from_secs(6);
const EXIT_TIMEOUT: Duration = Duration::from_secs(10);
const EXIT_LATENCY_TIMEOUT: Duration = Duration::from_secs(2);
const EXIT_LATENCY_BUDGET: Duration = Duration::from_millis(250);

#[test]
fn windows_attach_exit_emits_exited_banner() -> Result<(), Box<dyn Error>> {
    let binary = PathBuf::from(env!("CARGO_BIN_EXE_rmux"));
    let label = format!("win-exit-{}", std::process::id());
    let _guard = RmuxServerGuard::new(&binary, label.clone());

    run_rmux(
        &binary,
        &label,
        ["new-session", "-d", "-s", "exitcase", "cmd.exe", "/D", "/K"],
    )?;
    run_rmux(&binary, &label, ["set-option", "-g", "status", "off"])?;

    let mut attach = ChildCommand::new(&binary)
        .args(["-L", &label, "attach-session", "-t", "exitcase"])
        .size(TerminalSize::new(100, 30))
        .spawn()?;
    let io = attach.master().try_clone_io()?;

    wait_for_needle_or_error(&mut attach, b">", SETUP_TIMEOUT)?;
    io.write_all(b"echo RMUX_EXIT_READY\r\n")?;
    wait_for_needle_or_error(&mut attach, b"RMUX_EXIT_READY", SETUP_TIMEOUT)?;
    io.write_all(b"exit\r\n")?;

    let (exited, output) = wait_for_needle_or_terminate(&mut attach, b"[exited]", EXIT_TIMEOUT)?;
    terminate_spawned(&mut attach);
    assert!(
        exited,
        "attached Windows exit must print [exited]; observed output: {}",
        escaped_output(&output)
    );

    Ok(())
}

#[test]
#[ignore = "timing-sensitive Windows attach latency probe"]
fn windows_attach_exit_command_returns_under_latency_budget() -> Result<(), Box<dyn Error>> {
    let binary = PathBuf::from(env!("CARGO_BIN_EXE_rmux"));
    let label = format!("win-exit-latency-{}", std::process::id());
    let _guard = RmuxServerGuard::new(&binary, label.clone());

    run_rmux(
        &binary,
        &label,
        ["new-session", "-d", "-s", "exitcase", "cmd.exe", "/D", "/K"],
    )?;
    run_rmux(&binary, &label, ["set-option", "-g", "status", "off"])?;

    let mut attach = ChildCommand::new(&binary)
        .args(["-L", &label, "attach-session", "-t", "exitcase"])
        .size(TerminalSize::new(100, 30))
        .spawn()?;
    let io = attach.master().try_clone_io()?;

    wait_for_needle_or_error(&mut attach, b">", SETUP_TIMEOUT)?;
    let started = Instant::now();
    io.write_all(b"exit\r\n")?;
    let exited = wait_for_spawned_exit_or_terminate(&mut attach, EXIT_LATENCY_TIMEOUT)?;
    let elapsed = started.elapsed();

    assert!(
        exited,
        "attached Windows exit command did not return before timeout"
    );
    assert!(
        elapsed <= EXIT_LATENCY_BUDGET,
        "attached Windows exit command took {elapsed:?}, budget is {EXIT_LATENCY_BUDGET:?}"
    );

    Ok(())
}

#[test]
#[ignore = "timing-sensitive Windows attach latency probe"]
fn windows_attach_ctrl_d_returns_under_latency_budget_when_pwsh_available(
) -> Result<(), Box<dyn Error>> {
    if !pwsh_available() {
        eprintln!("skipping Ctrl-D latency probe because pwsh.exe is unavailable");
        return Ok(());
    }
    if !direct_pwsh_ctrl_d_exits()? {
        eprintln!(
            "skipping Ctrl-D latency probe because direct pwsh.exe ConPTY does not exit on the injected Ctrl-D"
        );
        return Ok(());
    }

    let binary = PathBuf::from(env!("CARGO_BIN_EXE_rmux"));
    let label = format!("win-ctrl-d-latency-{}", std::process::id());
    let _guard = RmuxServerGuard::new(&binary, label.clone());

    run_rmux(
        &binary,
        &label,
        [
            "new-session",
            "-d",
            "-s",
            "ctrldcase",
            "pwsh.exe",
            "-NoLogo",
            "-NoProfile",
        ],
    )?;
    run_rmux(&binary, &label, ["set-option", "-g", "status", "off"])?;

    let mut attach = ChildCommand::new(&binary)
        .args(["-L", &label, "attach-session", "-t", "ctrldcase"])
        .size(TerminalSize::new(100, 30))
        .spawn()?;

    wait_for_needle_or_error(&mut attach, b"PS ", SETUP_TIMEOUT)?;
    let started = Instant::now();
    write_windows_console_key(
        attach.child().pid(),
        WindowsConsoleKeyEvent::new(0x44, 0x20, 0x04, 0x0008, 1),
    )?;
    let exited = wait_for_spawned_exit_or_terminate(&mut attach, EXIT_LATENCY_TIMEOUT)?;
    let elapsed = started.elapsed();

    assert!(
        exited,
        "attached Windows Ctrl-D did not return before timeout"
    );
    assert!(
        elapsed <= EXIT_LATENCY_BUDGET,
        "attached Windows Ctrl-D took {elapsed:?}, budget is {EXIT_LATENCY_BUDGET:?}"
    );

    Ok(())
}

struct RmuxServerGuard<'a> {
    binary: &'a Path,
    label: String,
}

impl<'a> RmuxServerGuard<'a> {
    fn new(binary: &'a Path, label: String) -> Self {
        Self { binary, label }
    }
}

impl Drop for RmuxServerGuard<'_> {
    fn drop(&mut self) {
        let _ = Command::new(self.binary)
            .arg("-L")
            .arg(&self.label)
            .arg("kill-server")
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .status();
    }
}

fn run_rmux<const N: usize>(
    binary: &Path,
    label: &str,
    args: [&str; N],
) -> Result<(), Box<dyn Error>> {
    let status = Command::new(binary)
        .arg("-L")
        .arg(label)
        .args(args)
        .status()?;
    if !status.success() {
        return Err(io::Error::other(format!("rmux command failed with {status}")).into());
    }
    Ok(())
}

fn pwsh_available() -> bool {
    Command::new("pwsh.exe")
        .args([
            "-NoLogo",
            "-NoProfile",
            "-Command",
            "$PSVersionTable.PSVersion",
        ])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status()
        .is_ok_and(|status| status.success())
}

fn direct_pwsh_ctrl_d_exits() -> Result<bool, Box<dyn Error>> {
    let mut spawned = ChildCommand::new("pwsh.exe")
        .args(["-NoLogo", "-NoProfile"])
        .size(TerminalSize::new(100, 30))
        .spawn()?;

    wait_for_needle_or_error(&mut spawned, b"PS ", SETUP_TIMEOUT)?;
    write_windows_console_key(
        spawned.child().pid(),
        WindowsConsoleKeyEvent::new(0x44, 0x20, 0x04, 0x0008, 1),
    )?;
    wait_for_spawned_exit_or_terminate(&mut spawned, Duration::from_secs(1))
}

fn wait_for_needle_or_error(
    spawned: &mut SpawnedPty,
    needle: &[u8],
    timeout: Duration,
) -> Result<(), Box<dyn Error>> {
    let (found, output) = wait_for_needle_or_terminate(spawned, needle, timeout)?;
    if found {
        return Ok(());
    }
    Err(io::Error::new(
        io::ErrorKind::TimedOut,
        format!(
            "timed out waiting for {:?}; observed output: {}",
            String::from_utf8_lossy(needle),
            escaped_output(&output)
        ),
    )
    .into())
}

fn wait_for_needle_or_terminate(
    spawned: &mut SpawnedPty,
    needle: &[u8],
    timeout: Duration,
) -> Result<(bool, Vec<u8>), Box<dyn Error>> {
    let io = spawned.master().try_clone_io()?;
    let needle = needle.to_vec();
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let result = read_until_io(&io, &needle).map_err(|error| error.to_string());
        let _ = tx.send(result);
    });

    match rx.recv_timeout(timeout) {
        Ok(Ok(found)) => Ok(found),
        Ok(Err(error)) => Err(io::Error::other(error).into()),
        Err(mpsc::RecvTimeoutError::Timeout) => {
            terminate_spawned(spawned);
            match rx.recv_timeout(Duration::from_secs(2)) {
                Ok(Ok(result)) => Ok(result),
                Ok(Err(error)) => Err(io::Error::other(error).into()),
                Err(_) => Ok((false, Vec::new())),
            }
        }
        Err(mpsc::RecvTimeoutError::Disconnected) => {
            Err(io::Error::other("ConPTY reader thread disconnected").into())
        }
    }
}

fn read_until_io(io: &rmux_pty::PtyIo, needle: &[u8]) -> io::Result<(bool, Vec<u8>)> {
    let mut output = Vec::new();
    let mut buffer = [0_u8; 4096];
    loop {
        let bytes_read = match io.read(&mut buffer) {
            Ok(bytes_read) => bytes_read,
            Err(error) if error.kind() == io::ErrorKind::BrokenPipe => return Ok((false, output)),
            Err(error) => return Err(error),
        };
        if bytes_read == 0 {
            return Ok((false, output));
        }
        output.extend_from_slice(&buffer[..bytes_read]);
        if output.windows(needle.len()).any(|window| window == needle) {
            return Ok((true, output));
        }
    }
}

fn wait_for_spawned_exit_or_terminate(
    spawned: &mut SpawnedPty,
    timeout: Duration,
) -> Result<bool, Box<dyn Error>> {
    let deadline = Instant::now() + timeout;
    loop {
        if spawned.child_mut().try_wait()?.is_some() {
            return Ok(true);
        }
        if Instant::now() >= deadline {
            terminate_spawned(spawned);
            return Ok(false);
        }
        thread::sleep(Duration::from_millis(5));
    }
}

fn escaped_output(output: &[u8]) -> String {
    String::from_utf8_lossy(output)
        .chars()
        .flat_map(char::escape_default)
        .collect()
}

fn terminate_spawned(spawned: &mut SpawnedPty) {
    let _ = spawned.child().terminate_forcefully();
    let _ = spawned.child_mut().wait();
}