use std::path::Path;
use super::*;
use crate::protocol::{Key, Mods};
use crate::testutil::{
here, install_fake_notifier, now_ms, read_pid, sh_env, wait_until, write_executable,
write_rollout,
};
fn sup(rows: u16, cols: u16) -> Supervisor {
let mut s = Supervisor::new(rows, cols, 2000);
s.set_launch_context(LaunchContext::here());
s
}
fn spawn(s: &mut Supervisor, cmd: impl Into<String>, cwd: PathBuf) {
s.apply(Command::Spawn {
command: cmd.into(),
cwd,
group: None,
});
}
fn spawn_grouped(s: &mut Supervisor, cmd: impl Into<String>, cwd: PathBuf, group: &str) {
s.apply(Command::Spawn {
command: cmd.into(),
cwd,
group: Some(group.into()),
});
}
#[test]
fn scrollback_resolution_precedence_clamp_and_fallback() {
assert_eq!(effective_scrollback(None, None), 2000);
assert_eq!(effective_scrollback(None, Some("500")), 500);
assert_eq!(effective_scrollback(None, Some("0")), 0);
assert_eq!(effective_scrollback(None, Some("999999999")), 100_000);
assert_eq!(effective_scrollback(None, Some("garbage")), 2000);
assert_eq!(effective_scrollback(None, Some("-5")), 2000);
assert_eq!(effective_scrollback(Some(5000), Some("500")), 5000);
assert_eq!(effective_scrollback(Some(999_999_999), None), 100_000);
assert_eq!(effective_scrollback(Some(0), Some("500")), 0);
}
#[test]
fn session_config_groups_by_dir_in_spawn_order() {
let mut s = sup(24, 80);
spawn(&mut s, "a", here());
spawn(&mut s, "b", PathBuf::from("/tmp"));
spawn(&mut s, "c", here());
let cfg = s.session_config();
assert_eq!(
cfg[&path::abbreviate(&here())],
vec![
SessionEntry {
cmd: "a".into(),
group: None,
name: None,
},
SessionEntry {
cmd: "c".into(),
group: None,
name: None,
},
]
);
assert_eq!(
cfg["/tmp"],
vec![SessionEntry {
cmd: "b".into(),
group: None,
name: None,
}]
);
}
#[test]
fn tick_emits_snapshot_and_watched_screen() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
s.tick();
let evs = s.drain();
assert_eq!(evs.len(), 1, "only a Tasks snapshot while unwatched");
let id = match &evs[0] {
Event::Tasks(v) => {
assert_eq!(v.len(), 1);
v[0].id
}
_ => panic!("expected a Tasks snapshot"),
};
s.apply(Command::Watch { id: Some(id) });
s.tick();
let evs = s.drain();
assert!(evs.iter().any(|e| matches!(e, Event::Tasks(_))));
assert!(
evs.iter()
.any(|e| matches!(e, Event::Screen(sv) if sv.id == id)),
"watching a task should stream its Screen"
);
}
#[test]
fn watched_screen_not_resent_when_unchanged() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
let mut id = 0;
for _ in 0..5 {
s.tick();
for e in s.drain() {
if let Event::Tasks(v) = e
&& let Some(t) = v.first()
{
id = t.id;
}
}
std::thread::sleep(Duration::from_millis(20));
}
assert!(id != 0, "task never appeared");
s.apply(Command::Watch { id: Some(id) });
s.tick();
assert!(
s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"first watched tick sends a full screen"
);
s.tick();
assert!(
!s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"unchanged screen must not be resent"
);
}
#[test]
fn decset_1007_flip_resends_watched_screen() {
let dir = scratch("flip_1007");
let ready = dir.join("ready");
let flag = dir.join("flag");
let mut s = sup(24, 80);
let cmd = format!(
"printf '\\033[?1049h'; touch {r}; until [ -e {f} ]; do sleep 0.05; done; \
printf '\\033[?1007l'; sleep 30",
r = ready.display(),
f = flag.display()
);
let id = spawn_ready(&mut s, cmd, here(), &ready);
s.apply(Command::Watch { id: Some(id) });
let open = wait_until(Duration::from_secs(5), || {
s.tick();
s.drain()
.iter()
.any(|e| matches!(e, Event::Screen(sv) if sv.alt_screen && sv.alt_scroll))
});
assert!(open, "the gate-open screen never arrived");
std::fs::write(&flag, b"").unwrap();
let closed = wait_until(Duration::from_secs(5), || {
s.tick();
s.drain()
.iter()
.any(|e| matches!(e, Event::Screen(sv) if sv.alt_screen && !sv.alt_scroll))
});
assert!(closed, "the ?1007l flip never re-sent the screen");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn tick_flushes_a_stalled_sync_update() {
let mut s = sup(24, 80);
spawn(
&mut s,
"printf 'begin\\033[?2026hstalled'; sleep 30",
here(),
);
let mut preview = String::new();
wait_until(Duration::from_secs(5), || {
s.tick();
for e in s.drain() {
if let Event::Tasks(v) = e
&& let Some(t) = v.first()
{
preview = t.preview.text.clone();
}
}
preview.contains("stalled")
});
assert!(
preview.contains("stalled"),
"the stalled sync frame never flushed; preview: {preview:?}"
);
}
fn scratch(tag: &str) -> PathBuf {
crate::testutil::temp(&format!("sup_{tag}"))
}
fn spawn_ready(s: &mut Supervisor, command: String, cwd: PathBuf, ready: &Path) -> u64 {
spawn(s, command, cwd);
assert!(
wait_until(Duration::from_secs(5), || ready.exists()),
"task never signalled ready"
);
first_id(s)
}
fn first_id(s: &mut Supervisor) -> u64 {
s.tick();
match s.drain().first() {
Some(Event::Tasks(v)) => v[0].id,
_ => panic!("expected a Tasks snapshot"),
}
}
fn wait_for_lifecycle(
s: &mut Supervisor,
id: u64,
pred: impl Fn(crate::protocol::Lifecycle) -> bool,
) {
let ok = wait_until(Duration::from_secs(5), || {
s.tick();
s.drain().iter().any(|e| {
matches!(e, Event::Tasks(v)
if v.iter().any(|t| t.id == id && pred(t.lifecycle)))
})
});
assert!(ok, "task {id} never reached the expected lifecycle");
}
#[test]
fn kill_delivers_term_before_kill() {
use crate::protocol::Lifecycle;
let dir = scratch("term_first");
let (ready, trapped) = (dir.join("ready"), dir.join("trapped"));
let mut s = sup(24, 80);
let id = spawn_ready(
&mut s,
format!(
"trap 'echo t > {t}; exit 0' TERM; echo r > {r}; while :; do sleep 0.1; done",
t = trapped.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
s.apply(Command::Kill { id });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
assert!(trapped.exists(), "the TERM trap never ran");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn term_ignoring_task_escalates_to_kill() {
use crate::protocol::Lifecycle;
let dir = scratch("escalate");
let ready = dir.join("ready");
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(150));
let id = spawn_ready(
&mut s,
format!(
"trap '' TERM; echo r > {r}; while :; do sleep 0.1; done",
r = ready.display()
),
dir.clone(),
&ready,
);
s.apply(Command::Kill { id });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Failed);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn shutdown_returns_early_when_tasks_respect_term() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 300", here());
let t0 = Instant::now();
s.apply(Command::Shutdown);
assert!(
t0.elapsed() < Duration::from_secs(1),
"shutdown waited the full grace for a TERM-respecting task"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty()))
);
}
#[test]
fn shutdown_survives_a_child_that_never_reads_stdin() {
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(200));
spawn(&mut s, "sleep 300", here());
let id = first_id(&mut s);
s.apply(Command::Paste {
id,
bytes: b"x\n".repeat(1 << 19),
});
let t0 = Instant::now();
s.apply(Command::Shutdown);
assert!(
t0.elapsed() < Duration::from_secs(2),
"shutdown blocked behind a PTY write to a non-reading child"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty()))
);
}
#[test]
fn overfull_writer_queue_refuses_message_with_notice() {
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(200));
spawn(&mut s, "sleep 300", here());
let id = first_id(&mut s);
let big = b"x\n".repeat(4 << 20);
s.apply(Command::Input {
id,
bytes: big.clone(),
});
s.apply(Command::Input {
id,
bytes: big.clone(),
});
s.apply(Command::Input { id, bytes: big });
let evs = s.drain();
assert!(
evs.iter().any(|e| matches!(e, Event::Status(m)
if m.contains(&format!("task {id}")) && m.contains("8 MiB"))),
"no refusal notice for the overflowing message; got {evs:?}"
);
let t0 = Instant::now();
s.apply(Command::Shutdown);
assert!(
t0.elapsed() < Duration::from_secs(2),
"supervisor wedged after a writer-queue refusal"
);
}
#[test]
fn shutdown_is_bounded_by_grace() {
let dir = scratch("shutdown_bound");
let ready = dir.join("ready");
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(200));
spawn_ready(
&mut s,
format!(
"trap '' TERM; echo r > {r}; while :; do sleep 0.1; done",
r = ready.display()
),
dir.clone(),
&ready,
);
let t0 = Instant::now();
s.apply(Command::Shutdown);
let elapsed = t0.elapsed();
assert!(
elapsed < Duration::from_secs(2),
"shutdown took {elapsed:?}: not bounded by the 200 ms grace"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty()))
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn clear_watch_stops_screen_stream_and_resets_dedup() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
let id = first_id(&mut s);
s.apply(Command::Watch { id: Some(id) });
s.tick();
assert!(
s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"watching should stream a Screen"
);
s.clear_watch();
s.tick();
assert!(
!s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"a disconnected client's watch must not keep streaming"
);
s.apply(Command::Watch { id: Some(id) });
s.tick();
assert!(
s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"re-watch after clear_watch must resend the full screen"
);
}
#[test]
fn clear_watch_snaps_the_watched_task_live() {
let mut s = sup(6, 80);
spawn(&mut s, "seq 1 200; sleep 30", here());
let id = first_id(&mut s);
s.apply(Command::Watch { id: Some(id) });
let scrolled = wait_until(Duration::from_secs(5), || {
s.tick();
let _ = s.drain();
s.apply(Command::Scrollback {
id,
action: ScrollAction::Up(3),
});
s.tasks[0].scroll_offset() > 0
});
assert!(scrolled, "the task never accrued scrollback");
s.clear_watch();
assert_eq!(
s.tasks[0].scroll_offset(),
0,
"disconnect must return the watched task's viewport to live"
);
}
#[test]
fn rerun_replaces_finished_task_in_place() {
use crate::protocol::Lifecycle;
let dir = scratch("rerun");
let marker = dir.join("marker");
let mut s = sup(24, 80);
spawn(
&mut s,
format!("echo run >> {}", marker.display()),
dir.clone(),
);
let id = first_id(&mut s);
s.apply(Command::Tag { id, on: true });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.apply(Command::Restart { id });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
let runs = std::fs::read_to_string(&marker).unwrap().lines().count();
assert_eq!(runs, 2, "rerun must re-execute the command");
s.tick();
let tagged = s
.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.iter().any(|t| t.id == id && t.tagged)));
assert!(tagged, "rerun must carry the tag over");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn group_names_normalize_at_the_boundary() {
let n = |s: &str| normalize_group(Some(s.to_string()));
assert_eq!(normalize_group(None), None);
assert_eq!(n("\x1b[31mapi\x07"), Some("[31mapi".into()));
assert_eq!(n(" backend "), Some("backend".into()));
assert_eq!(n(" \t \x1b \x7f \u{9b} "), None);
assert_eq!(n(""), None);
assert_eq!(n(&"\u{e9}".repeat(80)), Some("\u{e9}".repeat(64)));
assert_eq!(n(&format!(" {} ", "x".repeat(64))), Some("x".repeat(64)));
assert_eq!(n("Unassigned"), None);
assert_eq!(n(" Unassigned "), None);
assert_eq!(n("unassigned"), Some("unassigned".into()));
assert_eq!(n("UNASSIGNED"), Some("UNASSIGNED".into()));
assert_eq!(n("Api"), Some("Api".into()));
}
#[test]
fn display_names_normalize_at_the_boundary() {
let n = |s: &str| normalize_label(Some(s.to_string()));
assert_eq!(normalize_label(None), None);
assert_eq!(n("\x1b[31mapi\x07"), Some("[31mapi".into()));
assert_eq!(n(" backend "), Some("backend".into()));
assert_eq!(n(" \t \x1b \x7f \u{9b} "), None);
assert_eq!(n(""), None);
assert_eq!(n(&"\u{e9}".repeat(80)), Some("\u{e9}".repeat(64)));
assert_eq!(n(&format!(" {} ", "x".repeat(64))), Some("x".repeat(64)));
assert_eq!(n("Unassigned"), Some("Unassigned".into()));
}
#[test]
fn set_group_round_trips_and_clears() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
let id = first_id(&mut s);
let group_of = |s: &mut Supervisor| -> Option<String> {
s.tick();
for e in s.drain() {
if let Event::Tasks(v) = e
&& let Some(t) = v.iter().find(|t| t.id == id)
{
return t.group.clone();
}
}
panic!("task {id} missing from the snapshot");
};
s.apply(Command::SetGroup {
id,
group: Some(" api ".into()),
});
assert_eq!(group_of(&mut s), Some("api".into()));
s.apply(Command::SetGroup { id, group: None });
assert_eq!(group_of(&mut s), None);
s.apply(Command::SetGroup {
id: 999,
group: Some("ghost".into()),
});
assert!(s.drain().is_empty(), "unknown-id SetGroup must stay silent");
assert_eq!(group_of(&mut s), None);
}
#[test]
fn set_name_round_trips_and_clears() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
let id = first_id(&mut s);
let name_of = |s: &mut Supervisor| -> Option<String> {
s.tick();
for e in s.drain() {
if let Event::Tasks(v) = e
&& let Some(t) = v.iter().find(|t| t.id == id)
{
return t.name.clone();
}
}
panic!("task {id} missing from the snapshot");
};
s.apply(Command::SetName {
id,
name: Some(" api \x1b[2J ".into()),
});
assert_eq!(name_of(&mut s), Some("api [2J".into()));
s.apply(Command::SetName {
id,
name: Some("Unassigned".into()),
});
assert_eq!(name_of(&mut s), Some("Unassigned".into()));
s.apply(Command::SetName { id, name: None });
assert_eq!(name_of(&mut s), None);
s.apply(Command::SetName {
id: 999,
name: Some("ghost".into()),
});
assert!(s.drain().is_empty(), "unknown-id SetName must stay silent");
assert_eq!(name_of(&mut s), None);
}
#[test]
fn spawn_carries_a_normalized_group_from_birth() {
let mut s = sup(24, 80);
spawn_grouped(&mut s, "sleep 30", here(), " ui\x1b[2J ");
s.tick();
match s.drain().first() {
Some(Event::Tasks(v)) => assert_eq!(v[0].group.as_deref(), Some("ui[2J")),
_ => panic!("expected a Tasks snapshot"),
}
}
#[test]
fn rerun_carries_the_group_over() {
use crate::protocol::Lifecycle;
let mut s = sup(24, 80);
spawn_grouped(&mut s, "true", here(), "infra");
let id = first_id(&mut s);
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.apply(Command::Restart { id });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.tick();
let carried = s.drain().iter().any(|e| {
matches!(e, Event::Tasks(v)
if v.iter().any(|t| t.id == id && t.group.as_deref() == Some("infra")))
});
assert!(carried, "rerun must carry the group over");
}
#[test]
fn rerun_carries_the_name_over() {
use crate::protocol::Lifecycle;
let mut s = sup(24, 80);
spawn(&mut s, "true", here());
let id = first_id(&mut s);
s.apply(Command::SetName {
id,
name: Some("smoke".into()),
});
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.apply(Command::Restart { id });
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.tick();
let carried = s.drain().iter().any(|e| {
matches!(e, Event::Tasks(v)
if v.iter().any(|t| t.id == id && t.name.as_deref() == Some("smoke")))
});
assert!(carried, "rerun must carry the name over");
}
#[test]
fn rerun_refuses_running_task_and_unknown_id() {
use crate::protocol::Lifecycle;
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
let id = first_id(&mut s);
s.apply(Command::Restart { id });
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.contains("still running"))),
"a running task must be refused"
);
s.tick();
let alive = s.drain().iter().any(|e| {
matches!(e, Event::Tasks(v) if v.iter().any(
|t| t.id == id && matches!(t.lifecycle, Lifecycle::Active | Lifecycle::Idle)
))
});
assert!(alive, "the refused task must keep running");
s.apply(Command::Restart { id: 999 });
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.contains("no task"))),
);
}
#[test]
fn rerun_watched_task_resends_screen() {
use crate::protocol::Lifecycle;
let mut s = sup(24, 80);
spawn(&mut s, "true", here());
let id = first_id(&mut s);
wait_for_lifecycle(&mut s, id, |l| l == Lifecycle::Ok);
s.apply(Command::Watch { id: Some(id) });
s.tick();
assert!(
s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"first watched tick sends a full screen"
);
s.apply(Command::Restart { id });
s.tick();
assert!(
s.drain().iter().any(|e| matches!(e, Event::Screen(_))),
"rerun of the watched task must resend the screen"
);
}
#[test]
fn resize_clamps_hostile_dimensions() {
let mut s = sup(24, 80);
spawn(&mut s, "sleep 30", here());
s.apply(Command::Resize { rows: 0, cols: 0 });
s.tick(); let _ = s.drain();
assert_eq!((s.rows, s.cols), (1, 1), "zero dims clamp to the floor");
s.apply(Command::Resize {
rows: u16::MAX,
cols: u16::MAX,
});
s.tick(); let _ = s.drain();
assert!(s.rows >= 1 && s.rows <= MAX_DIM);
assert!(s.cols >= 1 && s.cols <= MAX_DIM);
assert!(
u32::from(s.rows) * u32::from(s.cols) <= MAX_CELLS,
"accepted geometry {}x{} exceeds MAX_CELLS ({MAX_CELLS}): its \
worst-case Screen frame would not fit MAX_FRAME",
s.rows,
s.cols,
);
s.apply(Command::Resize {
rows: MAX_DIM,
cols: MAX_DIM,
});
assert_eq!(u32::from(s.rows), u32::from(MAX_DIM));
assert_eq!(u32::from(s.cols), MAX_CELLS / u32::from(MAX_DIM));
s.apply(Command::Resize {
rows: 67,
cols: 302,
});
assert_eq!((s.rows, s.cols), (67, 302));
}
#[test]
fn worst_case_screen_frame_fits_max_frame() {
use std::fmt::Write as _;
use alacritty_terminal::{
event::VoidListener,
index::{Column, Line},
term::{Config, test::TermSize},
vte::ansi::Processor,
};
use crate::{
ansi,
frame::{KIND_SCREEN, MAX_FRAME},
protocol::{ScreenView, encode_event},
};
const SGR_A: &str =
"\x1b[0;1;2;3;4:5;7;8;9;38;2;255;254;253;48;2;252;251;250;58;2;249;248;247m";
const SGR_B: &str =
"\x1b[0;1;2;3;4:5;7;8;9;38;2;155;154;153;48;2;152;151;150;58;2;149;148;147m";
let rows = usize::from(MAX_DIM);
let cols = (MAX_CELLS / u32::from(MAX_DIM)) as usize;
assert_eq!(rows * cols, MAX_CELLS as usize, "geometry covers the bound");
let mut input = String::with_capacity(rows * cols * 100);
for row in 1..=rows {
for col in 1..=cols {
let sgr = if (row * cols + col).is_multiple_of(2) {
SGR_A
} else {
SGR_B
};
let _ = write!(input, "\x1b[{row};{col}H{sgr} \x1b[{row};{col}H\t");
}
}
let mut term =
alacritty_terminal::Term::new(Config::default(), &TermSize::new(cols, rows), VoidListener);
let mut parser: Processor = Processor::new();
parser.advance(&mut term, input.as_bytes());
let probe = &term.grid()[Line(0)][Column(0)];
assert_eq!(probe.c, '\t', "cells must take the expensive tab path");
assert!(
probe.underline_color().is_some(),
"cells must carry an underline color"
);
let (formatted, cursor, hide) = ansi::formatted(&term);
let lines: Vec<String> = ansi::contents(&term).lines().map(str::to_string).collect();
let (kind, payload) = encode_event(&Event::Screen(ScreenView {
id: 1,
lines,
formatted,
cursor,
hide_cursor: hide,
wants_mouse: false,
alt_screen: false,
alt_scroll: false,
scrollback: 0,
}));
assert_eq!(kind, KIND_SCREEN);
let per_cell = payload.len() as f64 / MAX_CELLS as f64;
assert!(
per_cell >= 85.0,
"worst-case construction degenerated: {per_cell:.1} bytes/cell"
);
assert!(
payload.len() + payload.len() / 4 <= MAX_FRAME as usize,
"worst-case Screen frame no longer fits MAX_FRAME with 25 % \
headroom: ansi::formatted emits {per_cell:.1} bytes/cell, \
MAX_CELLS is {MAX_CELLS}, MAX_FRAME is {MAX_FRAME}; shrink \
MAX_CELLS, cheapen the serializer, or raise MAX_FRAME",
);
}
fn reap_until(
s: &mut Supervisor,
budget: Duration,
mut pred: impl FnMut(&mut Supervisor) -> bool,
) -> bool {
wait_until(budget, || {
s.reap();
pred(s)
})
}
fn hello_with_sh(s: &mut Supervisor, cwd: PathBuf) {
s.set_launch_context(LaunchContext { env: sh_env(), cwd });
}
#[test]
fn remove_sweeps_stragglers_of_an_exited_leader() {
use nix::sys::signal::kill;
let dir = scratch("remove_sweep");
let (spid, ready) = (dir.join("spid"), dir.join("ready"));
let mut s = sup(24, 80);
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"trap '' HUP; sleep 300 & echo $! > {sp}; echo r > {r}",
sp = spid.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
let straggler = read_pid(&spid);
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.iter().all(|t| t.id != id || t.finished.is_some())
}));
assert!(kill(straggler, None).is_ok(), "straggler should be alive");
s.apply(Command::Remove { id });
assert!(
reap_until(&mut s, Duration::from_secs(5), |_| kill(straggler, None)
.is_err()),
"Remove never swept the straggler"
);
assert!(
reap_until(&mut s, Duration::from_secs(5), |s| s.graveyard.is_empty()),
"graveyard entry was never collected"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn rerun_sweeps_stragglers_of_the_old_run() {
use nix::sys::signal::kill;
let dir = scratch("rerun_sweep");
let (spid, ready) = (dir.join("spid"), dir.join("ready"));
let mut s = sup(24, 80);
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"trap '' HUP; sleep 300 & echo $! > {sp}; echo r > {r}",
sp = spid.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
let old_straggler = read_pid(&spid);
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.iter().all(|t| t.id != id || t.finished.is_some())
}));
assert!(kill(old_straggler, None).is_ok());
s.apply(Command::Restart { id });
assert!(
reap_until(&mut s, Duration::from_secs(5), |_| kill(
old_straggler,
None
)
.is_err()),
"rerun never swept the old run's straggler"
);
assert!(s.tasks.iter().any(|t| t.id == id));
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn kill_escalation_reaches_term_ignoring_straggler_after_leader_exit() {
use nix::sys::signal::kill;
let dir = scratch("kill_escalate_straggler");
let (spid, ready) = (dir.join("spid"), dir.join("ready"));
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(150));
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"trap '' HUP; (trap '' TERM; exec sleep 300) & echo $! > {sp}; echo r > {r}",
sp = spid.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
let straggler = read_pid(&spid);
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.iter().all(|t| t.id != id || t.finished.is_some())
}));
s.apply(Command::Kill { id }); assert!(
reap_until(&mut s, Duration::from_secs(5), |_| kill(straggler, None)
.is_err()),
"reap-driven escalation never KILLed the straggler"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn shutdown_waits_for_graveyard_grace() {
use nix::sys::signal::kill;
let dir = scratch("shutdown_graveyard");
let (spid, ready) = (dir.join("spid"), dir.join("ready"));
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(400));
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"trap '' HUP; (trap '' TERM; exec sleep 300) & echo $! > {sp}; echo r > {r}",
sp = spid.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
let straggler = read_pid(&spid);
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.iter().all(|t| t.id != id || t.finished.is_some())
}));
s.apply(Command::Remove { id }); let alive_mid_grace = std::thread::spawn(move || {
std::thread::sleep(Duration::from_millis(200));
kill(straggler, None).is_ok()
});
s.apply(Command::Shutdown);
assert!(
alive_mid_grace.join().unwrap(),
"straggler was KILLed before its grace elapsed"
);
assert!(
reap_until(&mut s, Duration::from_secs(5), |_| kill(straggler, None)
.is_err()),
"straggler survived shutdown"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn shutdown_holds_the_grace_for_members_of_an_exited_leader() {
use nix::sys::signal::{Signal, kill};
let dir = scratch("shutdown_leaderless");
let (spid, ready) = (dir.join("spid"), dir.join("ready"));
let mut s = sup(24, 80);
s.set_kill_grace(Duration::from_millis(400));
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"trap '' HUP; (trap '' TERM; exec sleep 300) & echo $! > {sp}; echo r > {r}",
sp = spid.display(),
r = ready.display()
),
dir.clone(),
&ready,
);
let straggler = read_pid(&spid);
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.iter().all(|t| t.id != id || t.finished.is_some())
}));
assert!(kill(straggler, None).is_ok(), "straggler should be alive");
let t0 = Instant::now();
s.apply(Command::Shutdown);
let elapsed = t0.elapsed();
let survived = kill(straggler, None).is_ok();
let _ = kill(straggler, Signal::SIGKILL);
assert!(
elapsed >= Duration::from_millis(400),
"shutdown returned in {elapsed:?} with a non-empty group: the grace was skipped"
);
assert!(
elapsed < Duration::from_secs(2),
"shutdown took {elapsed:?}: not bounded by the 400 ms grace"
);
assert!(
survived,
"the straggler was KILLed instead of receiving the TERM grace"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn shutdown_is_prompt_when_every_group_is_already_empty() {
let mut s = sup(24, 80);
for _ in 0..2 {
spawn(&mut s, "true", here());
}
assert!(reap_until(&mut s, Duration::from_secs(5), |s| {
s.tasks.len() == 2 && s.tasks.iter().all(|t| t.finished.is_some())
}));
let t0 = Instant::now();
s.apply(Command::Shutdown);
assert!(
t0.elapsed() < Duration::from_millis(500),
"shutdown of already-empty groups took {:?}: the early exit is gone",
t0.elapsed()
);
}
#[test]
fn session_commands_use_the_launch_context_config_dir() {
let dir = scratch("sess_root");
let config = dir.join("config");
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(LaunchContext {
env: vec![(
"FLEETCOM_CONFIG_DIR".into(),
config.clone().into_os_string(),
)],
cwd: dir.clone(),
});
s.apply(Command::SaveSession { name: "ctx".into() });
assert!(
config.join("sessions").join("ctx.json").is_file(),
"save must land under the launch context's config dir"
);
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.starts_with("saved 'ctx'"))),
);
s.apply(Command::ListSessions);
let evs = s.drain();
assert!(
evs.iter()
.any(|e| matches!(e, Event::Sessions(n) if n == &["ctx".to_string()])),
"list must see the recipe save just wrote; got {evs:?}"
);
s.apply(Command::LoadSession { name: "ctx".into() });
let evs = s.drain();
assert!(
evs.iter()
.any(|e| matches!(e, Event::Status(m) if m.starts_with("loaded 'ctx'"))),
"load must find the recipe under the same root; got {evs:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn load_session_restores_saved_groups() {
let dir = scratch("sess_groups");
let config = dir.join("config");
let ctx = LaunchContext {
env: vec![(
"FLEETCOM_CONFIG_DIR".into(),
config.clone().into_os_string(),
)],
cwd: dir.clone(),
};
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(ctx.clone());
spawn_grouped(&mut s, "sleep 30", dir.clone(), "api");
spawn(&mut s, "sleep 31", dir.clone());
s.apply(Command::SaveSession {
name: "fleet".into(),
});
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.starts_with("saved 'fleet': 2 command(s)"))),
"save must still count commands"
);
let mut fresh = Supervisor::new(24, 80, 2000);
fresh.set_launch_context(ctx);
fresh.apply(Command::LoadSession {
name: "fleet".into(),
});
fresh.tick();
let evs = fresh.drain();
let tasks = evs
.iter()
.find_map(|e| match e {
Event::Tasks(v) => Some(v),
_ => None,
})
.expect("a Tasks snapshot after load");
let group_of = |cmd: &str| {
tasks
.iter()
.find(|t| t.command == cmd)
.unwrap_or_else(|| panic!("task '{cmd}' missing after load"))
.group
.clone()
};
assert_eq!(group_of("sleep 30"), Some("api".into()));
assert_eq!(group_of("sleep 31"), None);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn load_session_restores_saved_names() {
let dir = scratch("sess_names");
let config = dir.join("config");
let ctx = LaunchContext {
env: vec![(
"FLEETCOM_CONFIG_DIR".into(),
config.clone().into_os_string(),
)],
cwd: dir.clone(),
};
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(ctx.clone());
spawn(&mut s, "sleep 30", dir.clone());
spawn(&mut s, "sleep 31", dir.clone());
s.tick();
let id = match s.drain().first() {
Some(Event::Tasks(v)) => v.iter().find(|t| t.command == "sleep 30").unwrap().id,
_ => panic!("expected a Tasks snapshot"),
};
s.apply(Command::SetName {
id,
name: Some("api server".into()),
});
s.apply(Command::SaveSession {
name: "fleet".into(),
});
let mut fresh = Supervisor::new(24, 80, 2000);
fresh.set_launch_context(ctx);
fresh.apply(Command::LoadSession {
name: "fleet".into(),
});
fresh.tick();
let evs = fresh.drain();
let tasks = evs
.iter()
.find_map(|e| match e {
Event::Tasks(v) => Some(v),
_ => None,
})
.expect("a Tasks snapshot after load");
let name_of = |cmd: &str| {
tasks
.iter()
.find(|t| t.command == cmd)
.unwrap_or_else(|| panic!("task '{cmd}' missing after load"))
.name
.clone()
};
assert_eq!(name_of("sleep 30"), Some("api server".into()));
assert_eq!(name_of("sleep 31"), None);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn load_session_renormalizes_hand_edited_groups() {
let dir = scratch("sess_norm");
let config = dir.join("config");
std::fs::create_dir_all(config.join("sessions")).unwrap();
std::fs::write(
config.join("sessions").join("edited.json"),
format!(
r#"{{"{}": [{{"cmd": "sleep 30", "group": " x ", "name": " y "}}]}}"#,
dir.display()
),
)
.unwrap();
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(LaunchContext {
env: vec![("FLEETCOM_CONFIG_DIR".into(), config.into_os_string())],
cwd: dir.clone(),
});
s.apply(Command::LoadSession {
name: "edited".into(),
});
s.tick();
let evs = s.drain();
let restored = evs.iter().any(|e| {
matches!(e, Event::Tasks(v)
if v.iter().any(|t| t.group.as_deref() == Some("x")
&& t.name.as_deref() == Some("y")))
});
assert!(
restored,
"loaded group and name must come back normalized; got {evs:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
fn config_ctx(config: &Path, cwd: PathBuf, extra: &[(&str, &str)]) -> LaunchContext {
let mut env: Vec<(std::ffi::OsString, std::ffi::OsString)> = vec![(
"FLEETCOM_CONFIG_DIR".into(),
config.as_os_str().to_os_string(),
)];
for (k, v) in extra {
env.push(((*k).into(), (*v).into()));
}
LaunchContext { env, cwd }
}
#[test]
fn load_surfaces_parse_errors_instead_of_absence() {
let dir = scratch("sess_parse_err");
let config = dir.join("config");
std::fs::create_dir_all(config.join("sessions")).unwrap();
std::fs::write(config.join("sessions").join("broken.json"), "{not json").unwrap();
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(config_ctx(&config, dir.clone(), &[]));
s.apply(Command::LoadSession {
name: "broken".into(),
});
let evs = s.drain();
assert!(
evs.iter().any(
|e| matches!(e, Event::Status(m) if m.starts_with("session 'broken' failed to load:"))
),
"a parse failure must carry load_in's error; got {evs:?}"
);
assert!(
!evs.iter()
.any(|e| matches!(e, Event::Status(m) if m.contains("not found"))),
"a parse failure must not read as absence; got {evs:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn load_missing_session_reads_as_not_found() {
let dir = scratch("sess_missing");
let config = dir.join("config");
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(config_ctx(&config, dir.clone(), &[]));
s.apply(Command::LoadSession {
name: "ghost".into(),
});
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m == "session 'ghost' not found")),
"a missing recipe must still read as not found"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn load_reports_admit_failures_not_clean_success() {
let dir = scratch("sess_admit_fail");
let config = dir.join("config");
std::fs::create_dir_all(config.join("sessions")).unwrap();
std::fs::write(
config.join("sessions").join("fleet.json"),
format!(r#"{{"{}": ["true", "true"]}}"#, dir.display()),
)
.unwrap();
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(config_ctx(
&config,
dir.clone(),
&[("SHELL", "/nonexistent/no-such-shell")],
));
s.apply(Command::LoadSession {
name: "fleet".into(),
});
let evs = s.drain();
assert!(
evs.iter().any(|e| matches!(e, Event::Status(m)
if m.contains("2 failed to spawn") && !m.contains("task(s)"))),
"failed spawns must be reported, never folded into success; got {evs:?}"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty())),
"no task may exist when every spawn failed"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn spawn_refuses_over_length_command() {
let mut s = sup(24, 80);
s.apply(Command::Spawn {
command: "x".repeat(MAX_COMMAND_LEN + 1),
cwd: here(),
group: None,
});
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.contains("command too long"))),
"an over-length spawn must be refused with a notice"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty())),
"no task may exist after a refused spawn"
);
}
#[test]
fn load_skips_over_length_commands() {
let dir = scratch("sess_cmd_len");
let config = dir.join("config");
std::fs::create_dir_all(config.join("sessions")).unwrap();
std::fs::write(
config.join("sessions").join("big.json"),
format!(
r#"{{"{}": ["true", "{}"]}}"#,
dir.display(),
"x".repeat(MAX_COMMAND_LEN + 1)
),
)
.unwrap();
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(config_ctx(&config, dir.clone(), &[]));
s.apply(Command::LoadSession { name: "big".into() });
let evs = s.drain();
assert!(
evs.iter().any(|e| matches!(e, Event::Status(m)
if m.contains("1 task(s)") && m.contains("1 skipped"))),
"the over-length entry must be counted as skipped; got {evs:?}"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn spawn_uses_the_launch_context_env_not_the_process_env() {
assert!(
std::env::var_os("USER").is_some(),
"test needs USER set in the process env to prove it doesn't leak"
);
let dir = scratch("hello_env");
let out = dir.join("out");
let mut s = Supervisor::new(24, 80, 2000);
s.set_launch_context(LaunchContext {
env: vec![("FLEETCOM_MARKER".into(), "xyzzy".into())],
cwd: dir.clone(),
});
spawn(
&mut s,
format!(
"printf '%s:%s' \"$FLEETCOM_MARKER\" \"${{USER:-unset}}\" > {}",
out.display()
),
dir.clone(),
);
let ok = reap_until(&mut s, Duration::from_secs(5), |_| {
std::fs::read_to_string(&out).is_ok_and(|c| !c.is_empty())
});
assert!(ok, "the marker task never wrote its output");
assert_eq!(std::fs::read_to_string(&out).unwrap(), "xyzzy:unset");
let _ = std::fs::remove_dir_all(&dir);
}
#[test]
fn launch_without_context_is_refused() {
let mut s = Supervisor::new(24, 80, 2000);
spawn(&mut s, "true", here());
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Status(m) if m.contains("no launch context"))),
"context-less spawn must be refused with a status notice"
);
s.tick();
assert!(
s.drain()
.iter()
.any(|e| matches!(e, Event::Tasks(v) if v.is_empty())),
"no task may exist after a refused spawn"
);
s.apply(Command::LoadSession { name: "any".into() });
let evs = s.drain();
assert!(
evs.iter().any(|e| matches!(e, Event::Status(m)
if m.contains("no launch context") || m.contains("not found"))),
"context-less load must not spawn; got {evs:?}"
);
}
#[test]
fn key_command_encodes_against_live_cursor_mode() {
let dir = scratch("key_live_mode");
let (ready, out) = (dir.join("ready"), dir.join("out"));
let mut s = sup(24, 80);
hello_with_sh(&mut s, dir.clone());
let id = spawn_ready(
&mut s,
format!(
"stty raw 2>/dev/null; printf '\\033[?1h\\033[?1049h'; echo r > {r}; cat > {o}",
r = ready.display(),
o = out.display()
),
dir.clone(),
&ready,
);
let app_cursor_on = |s: &Supervisor| {
s.tasks
.iter()
.find(|t| t.id == id)
.is_some_and(|t| t.input_hints().1)
};
assert!(
wait_until(Duration::from_secs(5), || app_cursor_on(&s)),
"child never entered application-cursor mode"
);
s.apply(Command::Key {
id,
code: Key::Up,
mods: Mods::default(),
});
let up: &[u8] = b"\x1bOA";
assert!(
wait_until(Duration::from_secs(5), || {
std::fs::read(&out).is_ok_and(|b| b == up)
}),
"Up under app-cursor must arrive as SS3 ESC O A; got {:?}",
std::fs::read(&out)
);
s.apply(Command::Key {
id,
code: Key::Left,
mods: Mods {
alt: true,
..Mods::default()
},
});
let total: &[u8] = b"\x1bOA\x1b[1;3D";
assert!(
wait_until(Duration::from_secs(5), || {
std::fs::read(&out).is_ok_and(|b| b == total)
}),
"Alt+Left must force CSI 1;3D under app-cursor; got {:?}",
std::fs::read(&out)
);
s.apply(Command::Kill { id });
let _ = std::fs::remove_dir_all(&dir);
}
#[path = "supervisor_capture_tests.rs"]
mod capture;