use anyhow::{Context, Result};
use tokio::process::Command;
pub(crate) fn socket() -> &'static str {
if is_test_binary() { "ninox-test" } else { "ninox" }
}
fn is_test_binary() -> bool {
std::env::current_exe()
.ok()
.is_some_and(|p| p.components().any(|c| c.as_os_str() == "deps"))
}
fn parse_tmux_version(raw: &str) -> (u32, u32) {
let ver = raw.trim().strip_prefix("tmux ").unwrap_or(raw.trim());
let mut parts = ver.split(|c: char| !c.is_ascii_digit());
let major: u32 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0);
let minor: u32 = parts.next().and_then(|s| s.parse().ok()).unwrap_or(0);
(major, minor)
}
pub fn detected_version_sync() -> (u32, u32) {
std::process::Command::new("tmux")
.arg("-V")
.output()
.ok()
.filter(|o| o.status.success())
.map(|o| parse_tmux_version(&String::from_utf8_lossy(&o.stdout)))
.unwrap_or((0, 0))
}
fn supports_extended_keys_format((major, minor): (u32, u32)) -> bool {
(major, minor) >= (3, 5)
}
fn server_config_for_version(version: (u32, u32)) -> String {
let mut cfg = String::from("# Managed by ninox — rewritten on every app start. Do not edit.\n");
cfg.push_str("set -g default-terminal \"tmux-256color\"\n");
cfg.push_str("set -as terminal-features \"xterm*:RGB:usstyle:extkeys:hyperlinks\"\n");
cfg.push_str("set -s extended-keys always\n");
if supports_extended_keys_format(version) {
cfg.push_str("set -s extended-keys-format csi-u\n");
}
cfg.push_str("set -g history-limit 100000\n");
cfg.push_str("set -g status off\n");
cfg.push_str("set -s escape-time 0\n");
cfg.push_str("set -g window-size latest\n");
cfg.push_str("set -g allow-passthrough on\n");
cfg.push_str("set -g focus-events on\n");
cfg.push_str("set -g exit-empty off\n");
cfg
}
fn config_path() -> std::path::PathBuf {
let file = if is_test_binary() { "tmux-test.conf" } else { "tmux.conf" };
dirs::config_dir()
.unwrap_or_else(|| std::path::PathBuf::from("/tmp"))
.join("ninox")
.join(file)
}
pub fn write_server_config() -> Result<std::path::PathBuf> {
let path = config_path();
if let Some(dir) = path.parent() {
std::fs::create_dir_all(dir)?;
}
std::fs::write(&path, server_config_for_version(detected_version_sync()))?;
Ok(path)
}
fn socket_args() -> Vec<String> {
vec!["-L".into(), socket().into()]
}
pub async fn require_version() -> Result<()> {
let out = Command::new("tmux").arg("-V").output().await
.context("tmux not found — install tmux (brew install tmux / apt install tmux)")?;
let v = String::from_utf8_lossy(&out.stdout);
let ver = v.trim().strip_prefix("tmux ").unwrap_or(v.trim());
let version = parse_tmux_version(&v);
anyhow::ensure!(
version >= (3, 2),
"ninox requires tmux >= 3.2 for extended keyboard support; found {ver}"
);
if !supports_extended_keys_format(version) {
tracing::warn!(
"tmux {ver} detected — extended-keys-format csi-u requires tmux >= 3.5; \
Shift+Enter and other disambiguated keys may not reach apps correctly \
on this version"
);
}
Ok(())
}
async fn ensure_server_ready() {
static READY: tokio::sync::OnceCell<()> = tokio::sync::OnceCell::const_new();
READY.get_or_init(|| async {
if let Err(e) = write_server_config() {
tracing::warn!("failed to write tmux config: {e}");
}
let conf = config_path().display().to_string();
let _ = run_raw(&["-L", socket(), "-f", &conf, "start-server"]).await;
let _ = run_raw(&["-L", socket(), "source-file", &conf]).await;
}).await;
}
fn is_missing_session(e: &anyhow::Error) -> bool {
let msg = e.to_string();
msg.contains("can't find session")
|| msg.contains("session not found")
|| msg.contains("no server running")
|| msg.contains("no sessions")
|| msg.contains("no current target")
}
#[derive(Debug, Clone)]
pub struct TmuxSession {
pub id: String,
pub created_ms: i64,
pub pid: Option<u32>,
pub tty: Option<String>,
}
async fn run(args: &[&str]) -> Result<String> {
ensure_server_ready().await;
let prefix = socket_args();
let mut full: Vec<&str> = prefix.iter().map(String::as_str).collect();
full.extend_from_slice(args);
run_raw(&full).await
}
async fn run_default(args: &[&str]) -> Result<String> {
run_raw(args).await
}
async fn run_raw(args: &[&str]) -> Result<String> {
let out = Command::new("tmux")
.args(args)
.kill_on_drop(true)
.output()
.await
.context("tmux not found — install tmux (brew install tmux / apt install tmux)")?;
if !out.status.success() {
let stderr = String::from_utf8_lossy(&out.stderr);
anyhow::bail!("tmux {:?} failed: {}", args, stderr.trim());
}
Ok(String::from_utf8_lossy(&out.stdout).trim_end().to_string())
}
async fn run_session_scoped(args: &[&str]) -> Result<String> {
match run(args).await {
Err(e) if is_missing_session(&e) => run_default(args).await,
other => other,
}
}
async fn run_best_effort(args: &[&str]) -> String {
match run(args).await {
Ok(result) => result,
Err(e) => {
tracing::warn!("tmux {:?} failed (ignored): {}", args, e);
String::new()
}
}
}
async fn run_best_effort_default(args: &[&str]) -> String {
match run_default(args).await {
Ok(result) => result,
Err(e) => {
tracing::warn!("tmux (default socket) {:?} failed (ignored): {}", args, e);
String::new()
}
}
}
fn shell_quote(s: &str) -> String {
format!("'{}'", s.replace('\'', "'\\''"))
}
pub async fn create_session(
id: &str,
workspace: &str,
cmd: &str,
env: &[(&str, &str)],
) -> Result<()> {
anyhow::ensure!(
std::path::Path::new(workspace).is_dir(),
"workspace directory does not exist: {workspace}"
);
let mut env_pairs: Vec<String> = Vec::new();
for (k, v) in env {
anyhow::ensure!(!k.contains('='), "env key must not contain '=': {k}");
env_pairs.push(format!("{k}={v}"));
}
let mut extra: Vec<&str> = Vec::new();
for pair in &env_pairs {
extra.push("-e");
extra.push(pair.as_str());
}
let shell = std::env::var("SHELL").unwrap_or_else(|_| "/bin/bash".to_string());
let shell_cmd = format!("{shell} -l -c {}", shell_quote(cmd));
let mut base = vec!["new-session", "-d", "-s", id, "-x", "140", "-y", "50", "-c", workspace];
base.extend_from_slice(&extra);
base.push(&shell_cmd);
run(&base).await.map(|_| ())
}
pub async fn kill_session(id: &str) -> Result<()> {
match run_session_scoped(&["kill-session", "-t", id]).await {
Ok(_) => Ok(()),
Err(e) => {
if is_missing_session(&e) {
Ok(())
} else {
Err(e)
}
}
}
}
pub async fn has_session(id: &str) -> bool {
run_session_scoped(&["has-session", "-t", id]).await.is_ok()
}
pub async fn list_sessions() -> Result<Vec<TmuxSession>> {
const SEP: &str = "|";
let fmt = format!("#{{session_name}}{SEP}#{{session_created}}{SEP}#{{pane_pid}}{SEP}#{{pane_tty}}");
let ninox_raw = run_best_effort(&["list-sessions", "-F", &fmt]).await;
let default_raw = run_best_effort_default(&["list-sessions", "-F", &fmt]).await;
let mut seen = std::collections::HashSet::new();
let mut sessions = Vec::new();
for raw in [ninox_raw, default_raw] {
for line in raw.lines().filter(|l| !l.is_empty()) {
let mut cols = line.splitn(4, SEP);
let Some(id) = cols.next().map(str::to_string) else { continue };
if !seen.insert(id.clone()) {
continue;
}
let sec = cols.next().and_then(|s| s.parse::<i64>().ok()).unwrap_or(0);
let pid = cols.next().and_then(|s| s.parse::<u32>().ok());
let tty = cols.next().map(str::to_string).filter(|s| !s.is_empty());
sessions.push(TmuxSession { id, created_ms: sec * 1000, pid, tty });
}
}
Ok(sessions)
}
pub async fn get_pane_tty(id: &str) -> Result<Option<String>> {
let out = run_session_scoped(&["list-panes", "-t", id, "-F", "#{pane_tty}"]).await?;
Ok(out
.lines()
.next()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty()))
}
pub async fn pipe_pane(id: &str, dest_path: &str) -> Result<()> {
run_session_scoped(&["pipe-pane", "-t", id, &format!("cat > {}", shell_quote(dest_path))]).await?;
Ok(())
}
pub async fn attach_args(session_id: &str) -> Vec<String> {
let mut argv = vec!["tmux".to_string()];
if run(&["has-session", "-t", session_id]).await.is_ok() {
argv.extend(socket_args());
} else {
tracing::warn!(
"session {session_id} predates the ninox socket — attaching on the \
legacy default tmux server without the managed config (extended \
keys / resize guarantees are degraded until it terminates naturally)"
);
}
argv.extend(["attach-session", "-t", session_id].map(String::from));
argv
}
pub async fn history_size(session_id: &str) -> i64 {
run_session_scoped(&["display-message", "-p", "-t", session_id, "#{history_size}"])
.await
.ok()
.and_then(|s| s.trim().parse().ok())
.unwrap_or(0)
}
pub async fn capture_history(session_id: &str, start: i64, end: i64) -> Vec<u8> {
run_session_scoped(&[
"capture-pane", "-p", "-e", "-t", session_id,
"-S", &start.to_string(), "-E", &end.to_string(),
])
.await
.map(|s| s.into_bytes())
.unwrap_or_default()
}
const SEND_SUBMIT_DELAY_MS: u64 = 300;
const SEND_VERIFY_ATTEMPTS: u32 = 3;
const SEND_VERIFY_DELAY_MS: u64 = 500;
pub async fn send_keys(session_id: &str, text: &str) -> Result<()> {
run_session_scoped(&["send-keys", "-t", session_id, "-l", text]).await?;
tokio::time::sleep(std::time::Duration::from_millis(SEND_SUBMIT_DELAY_MS)).await;
run_session_scoped(&["send-keys", "-t", session_id, "Enter"]).await?;
for attempt in 0..=SEND_VERIFY_ATTEMPTS {
tokio::time::sleep(std::time::Duration::from_millis(SEND_VERIFY_DELAY_MS)).await;
if !message_stuck_at_prompt(&capture_visible_plain(session_id).await, text) {
return Ok(());
}
if attempt < SEND_VERIFY_ATTEMPTS {
run_session_scoped(&["send-keys", "-t", session_id, "Enter"]).await?;
}
}
anyhow::bail!(
"message to {session_id} is still unsubmitted at its input prompt \
after {SEND_VERIFY_ATTEMPTS} Enter retries"
)
}
async fn capture_visible_plain(session_id: &str) -> String {
run_session_scoped(&["capture-pane", "-p", "-t", session_id])
.await
.unwrap_or_default()
}
fn prompt_line_content(pane: &str) -> Option<&str> {
const PROMPT: char = '❯';
let line = pane.lines().rev().find(|l| l.contains(PROMPT))?;
let after = &line[line.rfind(PROMPT).unwrap() + PROMPT.len_utf8()..];
Some(after.trim().trim_end_matches('│').trim())
}
fn message_stuck_at_prompt(pane: &str, text: &str) -> bool {
let Some(content) = prompt_line_content(pane) else {
return false;
};
if content.is_empty() {
return false;
}
if content.starts_with("[Pasted text") {
return true;
}
let first_line = text.lines().next().unwrap_or("").trim();
if first_line.is_empty() {
return false;
}
if content.starts_with(first_line) {
return true;
}
const MIN_TRUNCATED_MATCH_CHARS: usize = 10;
content.chars().count() >= MIN_TRUNCATED_MATCH_CHARS && first_line.starts_with(content)
}
const IDLE_WAKE_NUDGE: &str = ".";
fn nudge_still_alone_at_prompt(pane: &str) -> bool {
prompt_line_content(pane) == Some(IDLE_WAKE_NUDGE)
}
pub async fn wake_idle_session(session_id: &str) -> Result<()> {
let pane = capture_visible_plain(session_id).await;
match prompt_line_content(&pane) {
Some("") => {} Some(IDLE_WAKE_NUDGE) => {
return run_session_scoped(&["send-keys", "-t", session_id, "Enter"]).await.map(|_| ());
}
_ => return Ok(()), }
run_session_scoped(&["send-keys", "-t", session_id, "-l", IDLE_WAKE_NUDGE]).await?;
tokio::time::sleep(std::time::Duration::from_millis(SEND_SUBMIT_DELAY_MS)).await;
if !nudge_still_alone_at_prompt(&capture_visible_plain(session_id).await) {
return Ok(());
}
run_session_scoped(&["send-keys", "-t", session_id, "Enter"]).await.map(|_| ())
}
pub async fn paste_buffer(session_id: &str, buf_name: &str, tmp_path: &str, bytes: &[u8]) -> Result<()> {
std::fs::write(tmp_path, bytes)?;
let result = run_session_scoped(&[
"load-buffer", "-b", buf_name, tmp_path, ";",
"paste-buffer", "-b", buf_name, "-t", session_id, "-d",
]).await;
let _ = std::fs::remove_file(tmp_path);
result.map(|_| ())
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::time::{sleep, Duration};
fn tmux_available() -> bool {
std::process::Command::new("tmux")
.args(["-V"])
.output()
.map(|o| o.status.success())
.unwrap_or(false)
}
fn unique_id() -> String {
static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
format!(
"test-{}-{n}",
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_millis()
)
}
#[test]
fn tests_resolve_to_the_isolated_socket_not_the_real_apps() {
assert!(is_test_binary(), "the test binary itself must be detected as a test binary");
assert_eq!(socket(), "ninox-test");
assert_ne!(socket(), "ninox");
}
#[tokio::test]
async fn create_session_fails_when_workspace_dir_is_missing() {
if !tmux_available() { return; }
let id = unique_id();
let result = create_session(&id, "/definitely/not/a/real/dir", "sleep 30", &[]).await;
assert!(result.is_err(), "missing workspace must be an error, not a silent $HOME fallback");
assert!(!has_session(&id).await, "no session may be left behind on failure");
}
#[tokio::test]
async fn create_and_has_and_kill() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "sleep 30", &[]).await.unwrap();
assert!(has_session(&id).await);
kill_session(&id).await.unwrap();
assert!(!has_session(&id).await);
}
#[tokio::test]
async fn list_includes_created() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "sleep 30", &[]).await.unwrap();
let sessions = list_sessions().await.unwrap();
assert!(sessions.iter().any(|s| s.id == id));
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn get_pane_tty_returns_dev_path() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "sleep 30", &[]).await.unwrap();
let tty = get_pane_tty(&id).await.unwrap();
assert!(tty.map(|t| t.starts_with("/dev/")).unwrap_or(false));
kill_session(&id).await.unwrap();
}
const REACTION: &str = "[Ninox] CI is failing on your PR (1/3 checks). Please fix the following:\n - build\n\nRun the failing checks locally, fix the issues, and push your changes.";
#[test]
fn stuck_when_pasted_text_marker_sits_at_prompt() {
let pane = "\
● Working on the auth refactor now.
╭──────────────────────────────────────────────╮
│ ❯ [Pasted text #1 +4 lines] │
╰──────────────────────────────────────────────╯
? for shortcuts";
assert!(message_stuck_at_prompt(pane, REACTION));
}
#[test]
fn stuck_when_message_text_sits_at_prompt() {
let pane = "\
╭──────────────────────────────────────────────────────╮
│ ❯ [Ninox] Worker `auth-fix`'s PR #12 merged. │
╰──────────────────────────────────────────────────────╯";
assert!(message_stuck_at_prompt(pane, "[Ninox] Worker `auth-fix`'s PR #12 merged."));
}
#[test]
fn stuck_when_prompt_shows_truncated_prefix_of_message() {
let pane = "│ ❯ [Ninox] CI is failing on your PR (1/3 che│";
assert!(message_stuck_at_prompt(pane, REACTION));
}
#[test]
fn not_stuck_when_prompt_empty_and_message_echoed_in_transcript() {
let pane = "\
❯ [Ninox] Worker `auth-fix`'s PR #12 merged.
● Noted — spawning the follow-up worker.
╭──────────────────────────────────────────────╮
│ ❯ │
╰──────────────────────────────────────────────╯";
assert!(!message_stuck_at_prompt(pane, "[Ninox] Worker `auth-fix`'s PR #12 merged."));
}
#[test]
fn not_stuck_when_prompt_holds_unrelated_text() {
let pane = "│ ❯ can you also update the readme while │";
assert!(!message_stuck_at_prompt(pane, REACTION));
}
#[test]
fn not_stuck_when_pane_has_no_prompt_marker() {
let pane = "$ echo hi\nhi\n$";
assert!(!message_stuck_at_prompt(pane, "echo hi"));
}
#[test]
fn not_stuck_when_prompt_holds_only_a_short_prefix_of_message() {
for typed in ["[", "[Ninox] C"] {
let pane = format!("│ ❯ {typed} │");
assert!(
!message_stuck_at_prompt(&pane, REACTION),
"short prompt content {typed:?} must not count as our stuck message",
);
}
}
#[test]
fn prompt_line_content_distinguishes_missing_from_empty_prompt() {
assert_eq!(prompt_line_content("$ echo hi\nhi\n$"), None);
assert_eq!(prompt_line_content("│ ❯ │"), Some(""));
assert_eq!(
prompt_line_content("│ ❯ hello there │"),
Some("hello there"),
);
}
#[test]
fn nudge_still_alone_at_prompt_true_when_only_the_nudge_is_present() {
let pane = "│ ❯ . │";
assert!(nudge_still_alone_at_prompt(pane));
}
#[test]
fn nudge_still_alone_at_prompt_false_once_more_text_is_appended() {
let pane = "│ ❯ .oops I started typing │";
assert!(!nudge_still_alone_at_prompt(pane));
}
#[test]
fn nudge_still_alone_at_prompt_false_once_the_prompt_is_empty_again() {
let pane = "│ ❯ │";
assert!(!nudge_still_alone_at_prompt(pane));
}
#[test]
fn nudge_still_alone_at_prompt_false_when_no_prompt_is_visible() {
assert!(!nudge_still_alone_at_prompt("$ echo hi\nhi\n$"));
}
#[tokio::test]
async fn wake_idle_session_skips_when_prompt_has_content() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "cat", &[]).await.unwrap();
sleep(Duration::from_millis(300)).await;
run_session_scoped(&["send-keys", "-t", &id, "-l", "❯ half-typed message"]).await.unwrap();
run_session_scoped(&["send-keys", "-t", &id, "Enter"]).await.unwrap();
wait_for_pane_contains(&id, "half-typed message").await;
wake_idle_session(&id).await.unwrap();
sleep(Duration::from_millis(300)).await;
let pane = capture_visible_plain(&id).await;
assert!(
!pane.lines().any(|l| l.trim() == IDLE_WAKE_NUDGE),
"must never nudge over a non-empty prompt: {pane}"
);
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn wake_idle_session_sends_nudge_when_prompt_is_empty() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "cat", &[]).await.unwrap();
sleep(Duration::from_millis(300)).await;
run_session_scoped(&["send-keys", "-t", &id, "-l", "❯ "]).await.unwrap();
wait_for_pane_contains(&id, "❯").await;
wake_idle_session(&id).await.unwrap();
wait_for_pane_contains(&id, "❯ .").await;
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn wake_idle_session_submits_its_own_stale_nudge_instead_of_stalling_forever() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "cat", &[]).await.unwrap();
sleep(Duration::from_millis(300)).await;
let marker_line = format!("❯ {IDLE_WAKE_NUDGE}");
run_session_scoped(&["send-keys", "-t", &id, "-l", &marker_line]).await.unwrap();
wait_for_pane_contains(&id, &marker_line).await;
wake_idle_session(&id).await.unwrap();
sleep(Duration::from_millis(300)).await;
let pane = capture_visible_plain(&id).await;
let occurrences = pane.lines().filter(|l| l.trim() == marker_line).count();
assert_eq!(
occurrences, 2,
"a stale nudge left over from a previous attempt must be submitted, \
not left stuck (which would wedge every future delivery): {pane}"
);
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn wake_idle_session_withholds_enter_if_typing_starts_during_the_delay() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "cat", &[]).await.unwrap();
sleep(Duration::from_millis(300)).await;
run_session_scoped(&["send-keys", "-t", &id, "-l", "❯ "]).await.unwrap();
wait_for_pane_contains(&id, "❯").await;
let id2 = id.clone();
let racer = tokio::spawn(async move {
sleep(Duration::from_millis(100)).await;
run_session_scoped(&["send-keys", "-t", &id2, "-l", "oops"]).await.unwrap();
});
wake_idle_session(&id).await.unwrap();
racer.await.unwrap();
sleep(Duration::from_millis(500)).await;
let pane = capture_visible_plain(&id).await;
let occurrences = pane.lines().filter(|l| l.contains("oops")).count();
assert_eq!(
occurrences, 1,
"the human's typing must never be submitted on our behalf: {pane}"
);
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn send_keys_delivers_to_a_plain_pane() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "cat", &[]).await.unwrap();
sleep(Duration::from_millis(300)).await;
send_keys(&id, "hello from ninox").await.unwrap();
let pane = capture_visible_plain(&id).await;
assert!(pane.contains("hello from ninox"), "pane: {pane}");
kill_session(&id).await.unwrap();
}
async fn wait_for_pane_contains(id: &str, needle: &str) {
for _ in 0..50 {
if capture_visible_plain(id).await.contains(needle) {
return;
}
sleep(Duration::from_millis(100)).await;
}
panic!("pane {id} never showed {needle:?}");
}
#[tokio::test]
async fn send_keys_errors_when_message_stays_stuck_at_a_prompt() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "printf '\\xe2\\x9d\\xaf '; exec cat > /dev/null", &[]).await.unwrap();
wait_for_pane_contains(&id, "❯").await;
let err = send_keys(&id, "stuck message").await.unwrap_err();
assert!(err.to_string().contains("unsubmitted"), "err: {err}");
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn send_keys_succeeds_when_the_last_retry_enter_submits() {
if !tmux_available() { return; }
let id = unique_id();
create_session(
&id, "/tmp",
"printf '\\xe2\\x9d\\xaf '; read a; read b; read c; read d; printf '\\x1b[2J\\x1b[H'; exec sleep 30",
&[],
).await.unwrap();
wait_for_pane_contains(&id, "❯").await;
send_keys(&id, "late message").await.unwrap();
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn send_keys_builds_correct_command() {
let quoted = shell_quote("hello world");
assert_eq!(quoted, "'hello world'");
let with_apostrophe = shell_quote("don't");
assert_eq!(with_apostrophe, "'don'\\''t'");
}
#[test]
fn server_config_is_written_and_contains_required_settings() {
let path = write_server_config().unwrap();
let body = std::fs::read_to_string(&path).unwrap();
for required in [
"default-terminal \"tmux-256color\"",
"extended-keys always",
"history-limit 100000",
"status off",
"window-size latest",
"allow-passthrough on",
"exit-empty off",
] {
assert!(body.contains(required), "config missing {required:?}\n{body}");
}
if supports_extended_keys_format(detected_version_sync()) {
assert!(body.contains("extended-keys-format csi-u"),
"config missing extended-keys-format on tmux >= 3.5\n{body}");
} else {
assert!(!body.contains("extended-keys-format"),
"config must omit extended-keys-format on tmux < 3.5 (rejected as invalid)\n{body}");
}
}
#[tokio::test]
async fn require_version_passes_on_installed_tmux() {
if !tmux_available() { return; }
require_version().await.unwrap();
}
#[tokio::test]
async fn sessions_are_created_on_the_ninox_socket() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "sleep 30", &[]).await.unwrap();
assert!(has_session(&id).await);
let default_out = std::process::Command::new("tmux")
.args(["has-session", "-t", &id])
.output()
.unwrap();
assert!(!default_out.status.success(), "session leaked onto the default socket");
kill_session(&id).await.unwrap();
}
#[tokio::test]
async fn legacy_default_socket_sessions_are_still_reachable() {
if !tmux_available() { return; }
let id = unique_id();
let st = std::process::Command::new("tmux")
.args(["new-session", "-d", "-s", &id, "-x", "80", "-y", "24", "sleep 30"])
.status()
.unwrap();
assert!(st.success());
assert!(has_session(&id).await, "has_session must fall back to the default socket");
let argv = attach_args(&id).await;
assert!(!argv.contains(&"-L".to_string()), "legacy session must attach without -L: {argv:?}");
kill_session(&id).await.unwrap(); assert!(!has_session(&id).await);
}
#[tokio::test]
async fn capture_history_returns_scrolled_off_lines() {
if !tmux_available() { return; }
let id = unique_id();
create_session(&id, "/tmp", "bash -c 'for i in $(seq 1 80); do echo line-$i; done; sleep 30'", &[]).await.unwrap();
sleep(Duration::from_millis(500)).await;
let hist = history_size(&id).await;
assert!(hist > 0, "expected history to accumulate, got {hist}");
let bytes = capture_history(&id, -hist, -1).await;
let text = String::from_utf8_lossy(&bytes);
assert!(text.contains("line-1"), "oldest line missing from history capture: {text}");
kill_session(&id).await.unwrap();
}
}