use crossterm::{
event::{DisableMouseCapture, EnableMouseCapture},
execute,
terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode},
};
use std::env;
use std::future::Future;
use std::io;
use std::io::Write as _;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::Stdio;
use tokio::fs;
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::process::Command;
use crate::events::{
OutputStream, UpgradeEvent, emit_upgrade_event, tui_active, upgrade_events_enabled,
};
pub async fn command_exists(name: &str) -> bool {
resolve_command_path(name).await.is_some()
}
pub async fn resolve_command_path(name: &str) -> Option<PathBuf> {
if command_name_is_path(name) {
return executable_path(name).await;
}
let candidates = command_name_candidates(name);
find_command_in_path(&candidates).await
}
fn command_name_is_path(name: &str) -> bool {
name.contains('/') || name.contains('\\')
}
async fn executable_path(name: &str) -> Option<PathBuf> {
let path = Path::new(name);
is_executable(path).await.then(|| path.to_path_buf())
}
async fn find_command_in_path(candidates: &[String]) -> Option<PathBuf> {
for dir in env::split_paths(&env::var_os("PATH")?) {
for candidate_name in candidates {
let candidate = dir.join(candidate_name);
if is_executable(&candidate).await {
return Some(candidate);
}
}
}
None
}
#[cfg(windows)]
fn command_name_candidates(name: &str) -> Vec<String> {
let path = Path::new(name);
if path.extension().is_some() {
return vec![name.to_string()];
}
let pathext = env::var_os("PATHEXT")
.and_then(|value| value.into_string().ok())
.unwrap_or_else(|| ".COM;.EXE;.BAT;.CMD".to_string());
let mut candidates = Vec::new();
for ext in pathext
.split(';')
.map(str::trim)
.filter(|ext| !ext.is_empty())
{
candidates.push(format!("{name}{ext}"));
}
candidates.push(name.to_string());
candidates
}
#[cfg(not(windows))]
fn command_name_candidates(name: &str) -> Vec<String> {
vec![name.to_string()]
}
async fn command_program(program: &str) -> PathBuf {
resolve_command_path(program)
.await
.unwrap_or_else(|| PathBuf::from(program))
}
async fn command(program: &str) -> Command {
let program_path = command_program(program).await;
#[cfg(windows)]
{
if program_path
.extension()
.and_then(|ext| ext.to_str())
.is_some_and(|ext| ext.eq_ignore_ascii_case("cmd") || ext.eq_ignore_ascii_case("bat"))
{
let mut cmd = Command::new("cmd.exe");
cmd.arg("/D").arg("/C").arg("call").arg(program_path);
return cmd;
}
}
Command::new(program_path)
}
async fn is_executable(path: &Path) -> bool {
let Ok(meta) = fs::metadata(path).await else {
return false;
};
if !meta.is_file() {
return false;
}
#[cfg(unix)]
{
meta.permissions().mode() & 0o111 != 0
}
#[cfg(not(unix))]
{
true
}
}
pub async fn run_capture(program: &str, args: &[&str]) -> io::Result<(i32, String)> {
let output = command(program).await.args(args).output().await?;
let code = output.status.code().unwrap_or(-1);
let mut text = String::new();
text.push_str(&String::from_utf8_lossy(&output.stdout));
text.push_str(&String::from_utf8_lossy(&output.stderr));
Ok((code, text))
}
pub async fn run_capture_streaming(program: &str, args: &[&str]) -> io::Result<(i32, String)> {
let mut child = command(program)
.await
.args(args)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let stdout = child
.stdout
.take()
.ok_or_else(|| io::Error::other("stdout pipe unavailable"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| io::Error::other("stderr pipe unavailable"))?;
let stdout_task = tokio::spawn(read_streaming_output(stdout));
let stderr_task = tokio::spawn(read_streaming_output(stderr));
let status = child.wait().await?;
let stdout_text = join_stream_reader(stdout_task).await?;
let stderr_text = join_stream_reader(stderr_task).await?;
let mut text = String::new();
text.push_str(&stdout_text);
text.push_str(&stderr_text);
Ok((status.code().unwrap_or(-1), text))
}
pub async fn run_capture_streaming_evented(
program: &str,
args: &[&str],
) -> io::Result<(i32, String)> {
emit_upgrade_event(UpgradeEvent::CommandStarted {
program: program.to_string(),
args: args.iter().map(|arg| (*arg).to_string()).collect(),
});
let mut child = command(program)
.await
.args(args)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let stdout = child
.stdout
.take()
.ok_or_else(|| io::Error::other("stdout pipe unavailable"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| io::Error::other("stderr pipe unavailable"))?;
let (stdout_result, stderr_result) = tokio::join!(
read_evented_output(stdout, OutputStream::Stdout),
read_evented_output(stderr, OutputStream::Stderr),
);
let status = child.wait().await?;
let stdout_text = stdout_result?;
let stderr_text = stderr_result?;
let mut text = String::new();
text.push_str(&stdout_text);
text.push_str(&stderr_text);
let code = status.code().unwrap_or(-1);
emit_upgrade_event(UpgradeEvent::CommandFinished {
success: status.success(),
code,
});
Ok((code, text))
}
async fn read_evented_output<R>(mut reader: R, stream: OutputStream) -> io::Result<String>
where
R: AsyncRead + Unpin + Send + 'static,
{
let mut bytes = Vec::new();
let mut buffer = [0; 8192];
loop {
let read = reader.read(&mut buffer).await?;
if read == 0 {
break;
}
let text = String::from_utf8_lossy(&buffer[..read]).into_owned();
emit_upgrade_event(UpgradeEvent::Output { stream, text });
bytes.extend_from_slice(&buffer[..read]);
}
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
async fn read_streaming_output<R>(mut reader: R) -> io::Result<String>
where
R: AsyncRead + Unpin + Send + 'static,
{
let mut bytes = Vec::new();
let mut buffer = [0; 8192];
loop {
let read = reader.read(&mut buffer).await?;
if read == 0 {
break;
}
write_stream_chunk(&buffer[..read])?;
bytes.extend_from_slice(&buffer[..read]);
}
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
fn write_stream_chunk(bytes: &[u8]) -> io::Result<()> {
let mut stdout = io::stdout().lock();
stdout.write_all(bytes)?;
stdout.flush()
}
async fn join_stream_reader(
handle: tokio::task::JoinHandle<io::Result<String>>,
) -> io::Result<String> {
handle
.await
.map_err(|err| io::Error::other(format!("stream reader task failed: {err}")))?
}
pub async fn run_inherit(program: &str, args: &[&str]) -> io::Result<bool> {
if upgrade_events_enabled() {
return run_capture_streaming_evented(program, args)
.await
.map(|(code, _)| code == 0);
}
run_inherit_direct(program, args).await
}
async fn run_inherit_direct(program: &str, args: &[&str]) -> io::Result<bool> {
let status = command(program)
.await
.args(args)
.stdin(Stdio::inherit())
.stdout(Stdio::inherit())
.stderr(Stdio::inherit())
.status()
.await?;
Ok(status.success())
}
pub async fn run_inherit_tty_handoff(
program: &str,
args: &[&str],
label: &str,
) -> io::Result<bool> {
if !tui_active() {
return run_inherit_direct(program, args).await;
}
emit_upgrade_event(UpgradeEvent::HandoffStarted {
label: label.to_string(),
});
suspend_tui_for_handoff()?;
let command_result = run_inherit_direct(program, args).await;
restore_tui_after_handoff()?;
let success = command_result.as_ref().is_ok_and(|result| *result);
emit_upgrade_event(UpgradeEvent::HandoffFinished { success });
command_result
}
pub async fn run_capture_streaming_tty_handoff(
program: &str,
args: &[&str],
label: &str,
) -> io::Result<(i32, String)> {
if !tui_active() {
return run_capture_streaming(program, args).await;
}
emit_upgrade_event(UpgradeEvent::HandoffStarted {
label: label.to_string(),
});
suspend_tui_for_handoff()?;
let command_result = run_capture_streaming_with_inherited_stdin(program, args).await;
restore_tui_after_handoff()?;
let success = command_result.as_ref().is_ok_and(|(code, _)| *code == 0);
emit_upgrade_event(UpgradeEvent::HandoffFinished { success });
command_result
}
async fn run_capture_streaming_with_inherited_stdin(
program: &str,
args: &[&str],
) -> io::Result<(i32, String)> {
let mut child = command(program)
.await
.args(args)
.stdin(Stdio::inherit())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let stdout = child
.stdout
.take()
.ok_or_else(|| io::Error::other("stdout pipe unavailable"))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| io::Error::other("stderr pipe unavailable"))?;
let stdout_task = tokio::spawn(read_streaming_output(stdout));
let stderr_task = tokio::spawn(read_streaming_output(stderr));
let status = child.wait().await?;
let stdout_text = join_stream_reader(stdout_task).await?;
let stderr_text = join_stream_reader(stderr_task).await?;
let mut text = String::new();
text.push_str(&stdout_text);
text.push_str(&stderr_text);
Ok((status.code().unwrap_or(-1), text))
}
fn suspend_tui_for_handoff() -> io::Result<()> {
disable_raw_mode()?;
execute!(std::io::stdout(), DisableMouseCapture, LeaveAlternateScreen)
}
fn restore_tui_after_handoff() -> io::Result<()> {
enable_raw_mode()?;
execute!(std::io::stdout(), EnterAlternateScreen, EnableMouseCapture)
}
pub async fn with_tui_handoff<F, T>(label: &str, future: F) -> io::Result<T>
where
F: Future<Output = T>,
{
if !tui_active() {
return Ok(future.await);
}
emit_upgrade_event(UpgradeEvent::HandoffStarted {
label: label.to_string(),
});
suspend_tui_for_handoff()?;
let result = future.await;
restore_tui_after_handoff()?;
emit_upgrade_event(UpgradeEvent::HandoffFinished { success: true });
Ok(result)
}
#[cfg(all(test, unix))]
mod tests {
use super::run_capture_streaming_evented;
use crate::events::{OutputStream, UpgradeEvent, with_upgrade_sink};
#[tokio::test]
async fn evented_command_merges_output_and_reports_exit_code() {
let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
let result = with_upgrade_sink(
Some(sender),
run_capture_streaming_evented(
"sh",
&["-c", "printf 'out\\n'; printf 'err\\n' >&2; exit 3"],
),
)
.await
.expect("command should spawn");
assert_eq!(result.0, 3);
let mut events = Vec::new();
while let Ok(event) = receiver.try_recv() {
events.push(event);
}
assert!(events.iter().any(|event| matches!(
event,
UpgradeEvent::CommandStarted { program, .. } if program == "sh"
)));
assert!(events.iter().any(|event| matches!(
event,
UpgradeEvent::Output { stream: OutputStream::Stdout, text } if text.contains("out")
)));
assert!(events.iter().any(|event| matches!(
event,
UpgradeEvent::Output { stream: OutputStream::Stderr, text } if text.contains("err")
)));
assert!(events.iter().any(|event| matches!(
event,
UpgradeEvent::CommandFinished {
success: false,
code: 3
}
)));
}
}
pub async fn run_cargo_install_update_capture(args: &[&str]) -> io::Result<(i32, String)> {
let mut proxy_args = vec!["install-update", "--locked"];
proxy_args.extend_from_slice(args);
run_capture("cargo-install-update", &proxy_args).await
}
pub async fn run_cargo_install_update_inherit(args: &[&str]) -> io::Result<bool> {
let mut proxy_args = vec!["install-update", "--locked"];
proxy_args.extend_from_slice(args);
run_inherit("cargo-install-update", &proxy_args).await
}