#![cfg(unix)]
mod common;
use std::error::Error;
use std::ffi::{OsStr, OsString};
use std::fs;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::OnceLock;
use std::time::Duration;
use rmux_proto::{
encode_frame, FrameDecoder, HasSessionRequest, ListSessionsRequest, Request, Response,
};
use rmux_sdk::bootstrap::discovery::{SDK_ENDPOINT_ENV, SDK_TIMEOUT_MS_ENV};
use rmux_sdk::{
EnsureSession, InfoSnapshot, NewSessionSpec, ProcessSpec, RmuxBuilder, RmuxError, SessionName,
TerminalSizeSpec,
};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::UnixStream;
use tokio::sync::Mutex;
use tokio::time::Instant;
type TestResult<T = ()> = Result<T, Box<dyn Error>>;
static LIVE_DAEMON_LOCK: common::unix_smoke::LiveDaemonLock =
common::unix_smoke::LiveDaemonLock::new();
static ENV_LOCK: Mutex<()> = Mutex::const_new(());
static UNIQUE_ID: AtomicUsize = AtomicUsize::new(0);
#[tokio::test]
async fn create_only_builds_live_session_and_hides_process_environment() -> TestResult {
let _lock = LIVE_DAEMON_LOCK.lock().await;
let harness = Harness::start("create-only").await?;
let rmux = harness.rmux();
let name = session_name("sdkcreate");
let secret = "SDK_SESSION_SECRET=do-not-render";
let ensure = EnsureSession::named(name.clone())
.create_only()
.environment([secret])
.empty_tags()
.window_name("main")
.size(TerminalSizeSpec::new(80, 24));
assert_eq!(ensure.configured_tags(), Some([].as_slice()));
assert_no_secret("ensure debug", &format!("{ensure:?}"), secret);
let session = ensure.ensure(&rmux).await?;
assert_eq!(session.name(), &name);
assert!(session.was_created());
assert_eq!(session.creation_tags(), Some([].as_slice()));
assert_no_secret("session debug", &format!("{session:?}"), secret);
assert!(session.exists().await?);
assert!(session.is_listed().await?);
let raw_has = framed_request(
harness.socket_path(),
Request::HasSession(HasSessionRequest {
target: name.clone(),
}),
)
.await?;
assert_eq!(
raw_has,
Response::HasSession(rmux_proto::HasSessionResponse { exists: true })
);
assert!(raw_list_session_names(harness.socket_path())
.await?
.contains(&name));
assert!(session.kill().await?);
harness.finish().await
}
#[tokio::test]
async fn ensure_session_command_preserves_legacy_and_explicit_modes_are_clear() -> TestResult {
let _lock = LIVE_DAEMON_LOCK.lock().await;
let harness = Harness::start("ensure-argv-shell").await?;
let rmux = harness.rmux();
let legacy_name = session_name("sdkensurelegacy");
let direct_name = session_name("sdkensureargv");
let shell_name = session_name("sdkensureshell");
let executable = write_executable_script(
harness.root(),
"ensure direct binary",
"#!/bin/sh\nprintf 'sdk_ensure_direct_argv\\n'; sleep 2\n",
)?;
let legacy = EnsureSession::named(legacy_name.clone())
.create_only()
.detached(true)
.command(["printf 'sdk_ensure_legacy_shell\\n'; sleep 2"])
.ensure(&rmux)
.await?;
legacy
.pane(0, 0)
.wait_for_text("sdk_ensure_legacy_shell")
.await?;
let direct = EnsureSession::named(direct_name.clone())
.create_only()
.detached(true)
.argv([executable.to_string_lossy().into_owned()])
.ensure(&rmux)
.await?;
direct
.pane(0, 0)
.wait_for_text("sdk_ensure_direct_argv")
.await?;
let shell = EnsureSession::named(shell_name.clone())
.create_only()
.detached(true)
.shell("printf 'sdk_ensure_shell\\n'; sleep 2")
.ensure(&rmux)
.await?;
shell.pane(0, 0).wait_for_text("sdk_ensure_shell").await?;
assert!(legacy.kill().await?);
assert!(direct.kill().await?);
assert!(shell.kill().await?);
harness.finish().await
}
#[tokio::test]
async fn create_only_duplicate_is_error_but_attach_if_exists_reuses() -> TestResult {
let _lock = LIVE_DAEMON_LOCK.lock().await;
let harness = Harness::start("reuse-policy").await?;
let rmux = harness.rmux();
let name = session_name("sdkreuse");
let created = EnsureSession::named(name.clone())
.create_only()
.ensure(&rmux)
.await?;
assert!(created.was_created());
let duplicate_secret = "DUPLICATE_SECRET=hidden";
let duplicate = EnsureSession::named(name.clone())
.create_only()
.environment([duplicate_secret])
.ensure(&rmux)
.await
.expect_err("create-only reports duplicate sessions");
assert_duplicate_session(&duplicate, "sdkreuse");
assert_no_secret(
"duplicate diagnostic",
&duplicate.to_string(),
duplicate_secret,
);
let reuse_secret = "REUSE_SECRET=hidden";
let reused = EnsureSession::named(name.clone())
.create_or_reuse()
.environment([reuse_secret])
.tag("kept")
.ensure(&rmux)
.await?;
assert_eq!(reused.name(), &name);
assert!(!reused.was_created());
assert_eq!(reused.creation_tags().expect("tag intent"), ["kept"]);
assert_no_secret("reused session debug", &format!("{reused:?}"), reuse_secret);
let fresh_name = session_name("sdkreusefresh");
let fresh = EnsureSession::named(fresh_name.clone())
.create_or_reuse()
.ensure(&rmux)
.await?;
assert_eq!(fresh.name(), &fresh_name);
assert!(fresh.was_created());
assert!(fresh.kill().await?);
let listed = raw_list_session_names(harness.socket_path()).await?;
assert_eq!(
listed
.iter()
.filter(|candidate| *candidate == &name)
.count(),
1
);
rmux.shutdown().await?;
harness.wait().await
}
#[tokio::test]
async fn reuse_only_policy_never_creates_sessions() -> TestResult {
let _lock = LIVE_DAEMON_LOCK.lock().await;
let harness = Harness::start("reuse-only").await?;
let rmux = harness.rmux();
let name = session_name("sdkreuseonly");
let missing = session_name("sdkmissing");
raw_new_session(harness.socket_path(), name.clone()).await?;
let reused = EnsureSession::named(name.clone())
.reuse_only()
.empty_tags()
.ensure(&rmux)
.await?;
assert_eq!(reused.name(), &name);
assert!(!reused.was_created());
assert_eq!(reused.creation_tags(), Some([].as_slice()));
let missing_error = EnsureSession::named(missing.clone())
.reuse_only()
.ensure(&rmux)
.await
.expect_err("reuse-only must not create a missing session");
assert_session_not_found(&missing_error, "sdkmissing");
assert!(!raw_list_session_names(harness.socket_path())
.await?
.contains(&missing));
let automatic_error = EnsureSession::auto_named()
.reuse_only()
.ensure(&rmux)
.await
.expect_err("reuse-only requires an explicit name");
assert!(
automatic_error
.to_string()
.contains("requires an explicit session name"),
"unexpected automatic-name reuse-only diagnostic: {automatic_error}"
);
rmux.shutdown().await?;
harness.wait().await
}
#[tokio::test]
async fn invalid_environment_override_diagnostic_is_redacted() -> TestResult {
let _lock = LIVE_DAEMON_LOCK.lock().await;
let harness = Harness::start("invalid-env").await?;
let rmux = harness.rmux();
let secret = "SDK_INVALID_ENV_SECRET";
let error = EnsureSession::named(session_name("sdkinvalidenv"))
.create_only()
.environment([secret])
.ensure(&rmux)
.await
.expect_err("daemon rejects invalid environment assignments");
let rendered = format!("{error:?}\n{error}");
assert_no_secret("invalid environment diagnostic", &rendered, secret);
assert!(
rendered.contains("[redacted process environment]"),
"diagnostic should identify redaction without exposing the override: {rendered}"
);
harness.finish().await
}
#[tokio::test]
async fn explicit_endpoint_timeout_and_empty_tag_semantics_are_preserved() -> TestResult {
let _live_lock = LIVE_DAEMON_LOCK.lock().await;
let _env_lock = ENV_LOCK.lock().await;
let harness = Harness::start("precedence").await?;
let wrong_endpoint = harness.root().join("wrong.sock");
let _endpoint = EnvGuard::set_os(SDK_ENDPOINT_ENV, wrong_endpoint.as_os_str());
let _timeout = EnvGuard::set(SDK_TIMEOUT_MS_ENV, "1");
let rmux = RmuxBuilder::new()
.unix_socket(harness.socket_path())
.default_timeout(Duration::from_secs(30))
.build();
let name = session_name("sdkprecedence");
let ensure = EnsureSession::named(name.clone())
.tag("")
.timeout(Duration::MAX);
assert_eq!(
rmux.endpoint(),
&rmux_sdk::RmuxEndpoint::UnixSocket(harness.socket_path().to_path_buf())
);
assert_eq!(ensure.configured_tags().expect("explicit tag"), [""]);
assert_eq!(ensure.resolved_timeout(&rmux), None);
let session = rmux.ensure_session(ensure).await?;
assert_eq!(session.endpoint(), rmux.endpoint());
assert_eq!(
session.configured_default_timeout(),
Some(Duration::from_secs(30))
);
assert_eq!(session.creation_tags().expect("explicit tag"), [""]);
assert!(session.exists().await?);
harness.finish().await
}
#[test]
fn process_related_debug_output_redacts_environment_values() {
let secret = "PROCESS_DEBUG_SECRET=hidden";
let process = ProcessSpec {
command: Some(vec!["printf".to_owned(), "ok".to_owned()]),
process_command: None,
environment: Some(vec![secret.to_owned()]),
};
assert_no_secret("process debug", &format!("{process:?}"), secret);
let spec = NewSessionSpec {
process: process.clone(),
..NewSessionSpec::default()
};
assert_no_secret("new-session spec debug", &format!("{spec:?}"), secret);
let ensure = EnsureSession::named(session_name("debugsecret")).process(process);
let ensure_debug = format!("{ensure:?}");
assert!(
ensure_debug.contains("printf"),
"ensure debug should keep non-secret process intent: {ensure_debug}"
);
assert_no_secret("ensure debug", &ensure_debug, secret);
let snapshot = InfoSnapshot::default();
assert_no_secret("info snapshot debug", &format!("{snapshot:?}"), secret);
}
fn session_name(value: &str) -> SessionName {
SessionName::new(value).expect("valid session name")
}
fn assert_duplicate_session(error: &RmuxError, expected: &str) {
match error {
RmuxError::Protocol {
source: rmux_proto::RmuxError::DuplicateSession(actual),
..
} => assert_eq!(actual, expected),
other => panic!("expected duplicate session diagnostic, got {other:?}"),
}
}
fn assert_session_not_found(error: &RmuxError, expected: &str) {
match error {
RmuxError::Protocol {
source: rmux_proto::RmuxError::SessionNotFound(actual),
..
} => assert_eq!(actual, expected),
other => panic!("expected session-not-found diagnostic, got {other:?}"),
}
}
fn assert_no_secret(label: &str, rendered: &str, secret: &str) {
assert!(
!rendered.contains(secret),
"{label} must not render process environment secrets: {rendered}",
);
if let Some((name, _value)) = secret.split_once('=') {
assert!(
!rendered.contains(name),
"{label} must not render process environment names: {rendered}",
);
}
assert!(
!rendered.contains("SECRET="),
"{label} must not render process environment bindings: {rendered}",
);
}
fn write_executable_script(root: &Path, name: &str, body: &str) -> TestResult<PathBuf> {
let path = root.join(name);
fs::write(&path, body)?;
let mut permissions = fs::metadata(&path)?.permissions();
permissions.set_mode(0o755);
fs::set_permissions(&path, permissions)?;
Ok(path)
}
async fn raw_list_session_names(socket_path: &Path) -> TestResult<Vec<SessionName>> {
match framed_request(
socket_path,
Request::ListSessions(ListSessionsRequest {
format: Some("#{session_name}".to_owned()),
filter: None,
sort_order: Some("name".to_owned()),
reversed: false,
}),
)
.await?
{
Response::ListSessions(response) => Ok(String::from_utf8_lossy(response.output.stdout())
.lines()
.map(SessionName::new)
.collect::<Result<Vec<_>, _>>()?),
response => Err(format!("unexpected list-sessions response: {response:?}").into()),
}
}
async fn raw_new_session(socket_path: &Path, name: SessionName) -> TestResult {
let request = NewSessionSpec {
session_name: Some(name.clone()),
detached: true,
..NewSessionSpec::default()
};
match framed_request(socket_path, Request::NewSessionExt(request.into())).await? {
Response::NewSession(response) => {
assert_eq!(response.session_name, name);
Ok(())
}
response => Err(format!("unexpected new-session response: {response:?}").into()),
}
}
async fn framed_request(socket_path: &Path, request: Request) -> TestResult<Response> {
let mut stream = UnixStream::connect(socket_path).await?;
let frame = encode_frame(&request)?;
stream.write_all(&frame).await?;
read_response(&mut stream).await
}
async fn read_response(stream: &mut UnixStream) -> TestResult<Response> {
let mut decoder = FrameDecoder::new();
let mut read_buffer = [0_u8; 8192];
loop {
if let Some(response) = decoder.next_frame::<Response>()? {
return Ok(response);
}
let bytes_read = stream.read(&mut read_buffer).await?;
if bytes_read == 0 {
return Err("connection closed before response frame".into());
}
decoder.push_bytes(&read_buffer[..bytes_read]);
}
}
struct Harness {
root: TestRoot,
socket_path: PathBuf,
child: Option<Child>,
}
impl Harness {
async fn start(label: &str) -> TestResult<Self> {
let root = TestRoot::new(label);
let socket_path = root.path().join("daemon.sock");
let mut child = Command::new(rmux_binary()?)
.arg("--__internal-daemon")
.arg(&socket_path)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()?;
wait_for_daemon_ready(&socket_path, &mut child).await?;
Ok(Self {
root,
socket_path,
child: Some(child),
})
}
fn root(&self) -> &Path {
self.root.path()
}
fn socket_path(&self) -> &Path {
&self.socket_path
}
fn rmux(&self) -> rmux_sdk::Rmux {
RmuxBuilder::new().unix_socket(&self.socket_path).build()
}
async fn wait(self) -> TestResult {
wait_for_child_exit(self, "server did not exit after shutdown request").await?;
Ok(())
}
async fn finish(self) -> TestResult {
let shutdown = self.rmux().shutdown().await;
wait_for_child_exit(self, "server did not exit during cleanup").await?;
if let Err(error) = shutdown {
let rendered = error.to_string();
assert!(
rendered.contains("connect to rmux daemon")
|| rendered.contains("rmux daemon closed the transport")
|| rendered.contains("rmux transport actor is closed")
|| rendered.contains("Connection reset by peer")
|| rendered.contains("Broken pipe"),
"unexpected cleanup shutdown error: {rendered}"
);
}
Ok(())
}
}
impl Drop for Harness {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() {
let _ = child.kill();
}
}
}
async fn wait_for_child_exit(mut harness: Harness, timeout_message: &'static str) -> TestResult {
let mut child = harness.child.take().expect("harness owns daemon child");
let deadline = Instant::now() + Duration::from_secs(60);
loop {
if let Some(status) = child.try_wait()? {
assert!(status.success(), "daemon exited with status {status}");
return Ok(());
}
if Instant::now() >= deadline {
let _ = child.kill();
return Err(timeout_message.into());
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
}
async fn wait_for_daemon_ready(socket_path: &Path, child: &mut Child) -> TestResult {
let deadline = Instant::now() + Duration::from_secs(60);
let probe = session_name("sdkprobe");
loop {
if let Some(status) = child.try_wait()? {
return Err(format!("daemon exited before accepting RPC: {status}").into());
}
if matches!(
framed_request(
socket_path,
Request::HasSession(HasSessionRequest {
target: probe.clone()
})
)
.await,
Ok(Response::HasSession(_))
) {
return Ok(());
}
if Instant::now() >= deadline {
return Err(format!(
"daemon at '{}' did not accept RPC before timeout",
socket_path.display()
)
.into());
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
}
fn rmux_binary() -> TestResult<&'static Path> {
static RMUX_BINARY: OnceLock<Result<PathBuf, String>> = OnceLock::new();
match RMUX_BINARY.get_or_init(|| resolve_rmux_binary().map_err(|error| error.to_string())) {
Ok(path) => Ok(path.as_path()),
Err(error) => Err(std::io::Error::other(error.clone()).into()),
}
}
fn resolve_rmux_binary() -> TestResult<PathBuf> {
if let Some(path) = option_env!("CARGO_BIN_EXE_rmux") {
let path = PathBuf::from(path);
if path.is_file() {
return Ok(path);
}
}
let target_dir = target_dir()?;
let candidate = target_dir.join("debug").join("rmux");
let status =
std::process::Command::new(std::env::var_os("CARGO").unwrap_or_else(|| "cargo".into()))
.arg("build")
.arg("--bin")
.arg("rmux")
.arg("--locked")
.arg("--manifest-path")
.arg(workspace_root().join("Cargo.toml"))
.env("CARGO_TARGET_DIR", &target_dir)
.status()?;
if !status.success() {
return Err(format!("failed to build rmux binary for daemon tests: {status}").into());
}
Ok(candidate)
}
fn target_dir() -> TestResult<PathBuf> {
if let Some(target_dir) = std::env::var_os("CARGO_TARGET_DIR") {
return Ok(PathBuf::from(target_dir));
}
let current = std::env::current_exe()?;
current
.parent()
.and_then(Path::parent)
.map(Path::to_path_buf)
.ok_or_else(|| "test executable is not under a target directory".into())
}
fn workspace_root() -> PathBuf {
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
.parent()
.and_then(Path::parent)
.expect("rmux-sdk manifest lives under crates/rmux-sdk")
.to_path_buf()
}
struct TestRoot {
path: PathBuf,
}
impl TestRoot {
fn new(label: &str) -> Self {
let unique_id = UNIQUE_ID.fetch_add(1, Ordering::Relaxed);
let path = PathBuf::from("/tmp").join(format!(
"rmux-sdk-session-{}-{}-{unique_id}",
compact_label(label),
std::process::id()
));
Self { path }
}
fn path(&self) -> &Path {
&self.path
}
}
impl Drop for TestRoot {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.path);
}
}
fn compact_label(label: &str) -> String {
let compact = label
.chars()
.filter(|character| character.is_ascii_alphanumeric())
.take(16)
.collect::<String>();
if compact.is_empty() {
"x".to_owned()
} else {
compact
}
}
struct EnvGuard {
key: &'static str,
previous: Option<OsString>,
}
impl EnvGuard {
fn set(key: &'static str, value: &str) -> Self {
Self::set_os(key, OsStr::new(value))
}
fn set_os(key: &'static str, value: &OsStr) -> Self {
let previous = std::env::var_os(key);
std::env::set_var(key, value);
Self { key, previous }
}
}
impl Drop for EnvGuard {
fn drop(&mut self) {
match self.previous.as_ref() {
Some(value) => std::env::set_var(self.key, value),
None => std::env::remove_var(self.key),
}
}
}