fix(cli): stop continuous groups on peer closure

Record a continuous peer-closed TargetFinished event as the reason for requesting group shutdown and stop each CLI/TUI group at most once. Drain the managed subscription and join before applying exit policy so queued output and target failures are preserved.

Give peer closure precedence over the cancellation caused by that internal stop while keeping explicit interruption and finite-run behavior distinct. Add bounded process regressions for partial closure, mixed failures, queued rows, and interruption plus focused TUI policy coverage.

crates/irtt-cli/src/cmd/client/run.rs

@@ -24,7 +24,10 @@ use super::{
 use crate::shared::client::expected_probe_count;
 use crate::shared::client::{
     is_shutdown_requested,
-    session::{peer_close_run_error, should_print_final_summary},
+    session::{
+        peer_close_run_error, request_group_stop_once, should_print_final_summary,
+        should_stop_group_on_peer_close,
+    },
     ClientSession,
 };
 
@@ -326,6 +329,8 @@ fn run_group_stream(
     };
 
     let mut interrupted = false;
+    let mut stop_requested = false;
+    let mut peer_close_requested_stop = false;
     let mut terminal_targets = HashSet::new();
     let mut last_event_at = Instant::now();
     let mut saw_target_event = false;
@@ -333,8 +338,10 @@ fn run_group_stream(
     loop {
         if is_shutdown_requested(shutdown_requested) {
             interrupted = true;
+            if request_group_stop_once(&mut stop_requested) {
                 session.stop();
             }
+        }
 
         match events.try_recv() {
             Ok(Some(group_event)) => {
@@ -362,6 +369,16 @@ fn run_group_stream(
                     ManagedGroupEvent::TargetFinished(target) => {
                         report_target_failure(&target);
                         terminal_targets.insert(target.id.as_str().to_owned());
+                        if should_stop_group_on_peer_close(
+                            continuous,
+                            interrupted,
+                            &target.end_reason,
+                        ) {
+                            peer_close_requested_stop = true;
+                            if request_group_stop_once(&mut stop_requested) {
+                                session.stop();
+                            }
+                        }
                     }
                 }
             }
@@ -422,12 +439,18 @@ fn run_group_stream(
     }
     let successful_targets = outcome.successful_target_outcomes;
     let failed_targets = outcome.failed_target_outcomes;
-    let terminal_error = if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted {
-        Some("managed client group was cancelled".to_owned())
-    } else if let Some(error) =
-        peer_close_run_error(continuous, interrupted, outcome.peer_closed_target_outcomes)
+    let peer_closed_target_outcomes = outcome
+        .peer_closed_target_outcomes
+        .max(u64::from(peer_close_requested_stop));
+    let terminal_error = if let Some(error) =
+        peer_close_run_error(continuous, interrupted, peer_closed_target_outcomes)
     {
         Some(error)
+    } else if outcome.end_reason == ManagedGroupEndReason::Cancelled
+        && !interrupted
+        && !peer_close_requested_stop
+    {
+        Some("managed client group was cancelled".to_owned())
     } else if !interrupted && successful_targets == 0 && failed_targets > 0 {
         Some(format!(
             "no managed target completed successfully ({failed_targets} failed)"

crates/irtt-cli/src/cmd/tui/run.rs

@@ -14,7 +14,11 @@ use irtt_client::{
 
 use crate::{
     cmd::tui::args::{ResolvedTuiTarget, TuiArgs},
-    shared::client::{is_shutdown_requested, session::peer_close_run_error, ClientSession},
+    shared::client::{
+        is_shutdown_requested,
+        session::{peer_close_run_error, request_group_stop_once, should_stop_group_on_peer_close},
+        ClientSession,
+    },
 };
 
 use super::ui::{should_render, TuiConfig, TuiState, TuiStatus, TuiTerminal};
@@ -177,6 +181,8 @@ fn run_group_tui(
     )?;
 
     let mut interrupted = false;
+    let mut stop_requested = false;
+    let mut peer_close_requested_stop = false;
     let mut terminal_targets = std::collections::HashSet::new();
     let mut saw_target_event = false;
     let mut last_event_at = Instant::now();
@@ -184,16 +190,20 @@ fn run_group_tui(
     let exit = loop {
         if is_shutdown_requested(shutdown_requested) {
             interrupted = true;
+            if request_group_stop_once(&mut stop_requested) {
                 session.stop();
             }
+        }
 
         if handle_input(state, shutdown_requested)? {
             render_if_due(terminal, state, next_render, true)?;
         }
         if state.quit_requested {
             interrupted = true;
+            if request_group_stop_once(&mut stop_requested) {
                 session.stop();
             }
+        }
 
         match events.try_recv() {
             Ok(Some(group_event)) => {
@@ -206,6 +216,16 @@ fn run_group_tui(
                     ManagedGroupEvent::TargetFinished(target) => {
                         terminal_targets.insert(target.id.as_str().to_owned());
                         state.process_target_outcome(&target);
+                        if should_stop_group_on_peer_close(
+                            args.is_continuous(),
+                            interrupted,
+                            &target.end_reason,
+                        ) {
+                            peer_close_requested_stop = true;
+                            if request_group_stop_once(&mut stop_requested) {
+                                session.stop();
+                            }
+                        }
                     }
                 }
                 render_if_due(terminal, state, next_render, false)?;
@@ -229,7 +249,7 @@ fn run_group_tui(
         }
     };
 
-    if exit.should_stop_before_join() {
+    if exit.should_stop_before_join() && request_group_stop_once(&mut stop_requested) {
         session.stop();
     }
 
@@ -260,7 +280,22 @@ fn run_group_tui(
     }
 
     interrupted |= is_shutdown_requested(shutdown_requested);
-    if outcome.end_reason == ManagedGroupEndReason::Cancelled && !interrupted {
+    let peer_closed_target_outcomes = outcome
+        .peer_closed_target_outcomes
+        .max(u64::from(peer_close_requested_stop));
+    if let Some(error) = peer_close_run_error(
+        args.is_continuous(),
+        interrupted,
+        peer_closed_target_outcomes,
+    ) {
+        state.set_run_error(error.clone());
+        render_if_due(terminal, state, next_render, true)?;
+        return Err(error.into());
+    }
+    if outcome.end_reason == ManagedGroupEndReason::Cancelled
+        && !interrupted
+        && !peer_close_requested_stop
+    {
         return match exit {
             GroupLoopExit::IdleGraceElapsed => {
                 Err("managed client group stayed idle before all targets completed".into())
@@ -273,15 +308,6 @@ fn run_group_tui(
             }
         };
     }
-    if let Some(error) = peer_close_run_error(
-        args.is_continuous(),
-        interrupted,
-        outcome.peer_closed_target_outcomes,
-    ) {
-        state.set_run_error(error.clone());
-        render_if_due(terminal, state, next_render, true)?;
-        return Err(error.into());
-    }
     let successful_targets = outcome.successful_target_outcomes;
     let failed_targets = outcome.failed_target_outcomes;
     if !interrupted && successful_targets == 0 && failed_targets > 0 {
@@ -507,4 +533,23 @@ mod tests {
         assert!(GroupLoopExit::SubscriptionDisconnected.should_stop_before_join());
         assert!(!GroupLoopExit::AllTargetsTerminal.should_stop_before_join());
     }
+
+    #[test]
+    fn continuous_group_peer_close_requests_stop_unless_interrupted() {
+        assert!(should_stop_group_on_peer_close(
+            true,
+            false,
+            &irtt_client::ManagedTargetEndReason::PeerClosed
+        ));
+        assert!(!should_stop_group_on_peer_close(
+            false,
+            false,
+            &irtt_client::ManagedTargetEndReason::PeerClosed
+        ));
+        assert!(!should_stop_group_on_peer_close(
+            true,
+            true,
+            &irtt_client::ManagedTargetEndReason::PeerClosed
+        ));
+    }
 }

crates/irtt-cli/src/shared/client/session.rs

@@ -4,7 +4,9 @@ use std::{
     time::{Duration, Instant},
 };
 
-use irtt_client::{Client, ClientConfig, ClientEvent, OpenOutcome, RecvBudget};
+use irtt_client::{
+    Client, ClientConfig, ClientEvent, ManagedTargetEndReason, OpenOutcome, RecvBudget,
+};
 
 pub const RECV_BUDGET: RecvBudget = RecvBudget { max_packets: 16 };
 const MAX_FINAL_DRAIN: Duration = Duration::from_secs(30);
@@ -162,6 +164,22 @@ pub fn peer_close_run_error(
     ))
 }
 
+pub fn should_stop_group_on_peer_close(
+    continuous: bool,
+    interrupted: bool,
+    end_reason: &ManagedTargetEndReason,
+) -> bool {
+    continuous && !interrupted && matches!(end_reason, ManagedTargetEndReason::PeerClosed)
+}
+
+pub fn request_group_stop_once(stop_requested: &mut bool) -> bool {
+    if *stop_requested {
+        return false;
+    }
+    *stop_requested = true;
+    true
+}
+
 pub fn final_drain_duration(probe_timeout: Duration) -> Duration {
     probe_timeout.min(MAX_FINAL_DRAIN)
 }
@@ -239,6 +257,34 @@ mod tests {
         );
     }
 
+    #[test]
+    fn grouped_peer_close_stop_policy_is_run_mode_and_interruption_aware() {
+        assert!(should_stop_group_on_peer_close(
+            true,
+            false,
+            &ManagedTargetEndReason::PeerClosed
+        ));
+        assert!(!should_stop_group_on_peer_close(
+            false,
+            false,
+            &ManagedTargetEndReason::PeerClosed
+        ));
+        assert!(!should_stop_group_on_peer_close(
+            true,
+            true,
+            &ManagedTargetEndReason::PeerClosed
+        ));
+        assert!(!should_stop_group_on_peer_close(
+            true,
+            false,
+            &ManagedTargetEndReason::TestComplete
+        ));
+
+        let mut stop_requested = false;
+        assert!(request_group_stop_once(&mut stop_requested));
+        assert!(!request_group_stop_once(&mut stop_requested));
+    }
+
     #[test]
     fn peer_close_followed_by_session_cleanup_is_successful() {
         let params = test_params(None, Duration::from_millis(10));

crates/irtt-cli/tests/client_multi_target.rs

@@ -4,6 +4,7 @@ use std::{
     io::Read,
     net::{SocketAddr, UdpSocket},
     process::{Command, Output, Stdio},
+    sync::mpsc,
     thread,
     thread::JoinHandle,
     time::{Duration, Instant},
@@ -29,6 +30,24 @@ impl FakeServer {
     }
 }
 
+struct InterruptibleFakeServer {
+    addr: SocketAddr,
+    opened: mpsc::Receiver<()>,
+    done: JoinHandle<()>,
+}
+
+impl InterruptibleFakeServer {
+    fn wait_until_opened(&self) {
+        self.opened
+            .recv_timeout(Duration::from_secs(2))
+            .expect("timed out waiting for CLI to open target");
+    }
+
+    fn join(self) {
+        self.done.join().unwrap();
+    }
+}
+
 #[test]
 fn list_columns_succeeds_without_target() {
     let output = Command::new(env!("CARGO_BIN_EXE_irtt-cli"))
@@ -337,10 +356,11 @@ fn continuous_all_peer_closed_targets_exit_nonzero() {
 }
 
 #[test]
-fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() {
+fn continuous_partial_peer_close_stops_promptly_and_preserves_queued_rows() {
     let params = test_params(None, Duration::from_millis(10));
-    let peer_closed = start_peer_close_server(params.clone());
-    let failure = start_open_failure_server(params);
+    let (healthy_reply_tx, healthy_reply_rx) = mpsc::channel();
+    let healthy = start_echo_server_with_first_reply(params.clone(), healthy_reply_tx);
+    let peer_closed = start_gated_peer_close_server(params, vec![healthy_reply_rx]);
     let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
     command.args([
         "--duration",
@@ -354,14 +374,79 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() {
         "--header",
         "never",
         "--target",
+        &format!("healthy={}", healthy.addr),
+        "--target",
         &format!("peer={}", peer_closed.addr),
+    ]);
+
+    let started_at = Instant::now();
+    let output = run_with_timeout(command, Duration::from_secs(2));
+    let elapsed = started_at.elapsed();
+    healthy.join();
+    peer_closed.join();
+
+    assert!(
+        !output.status.success(),
+        "status={:?}\nstdout={}",
+        output.status.code(),
+        String::from_utf8_lossy(&output.stdout)
+    );
+    assert!(
+        elapsed < Duration::from_secs(1),
+        "partial peer-close shutdown took {elapsed:?}"
+    );
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    let stderr = String::from_utf8(output.stderr).unwrap();
+    assert!(
+        stdout.lines().any(|line| line.starts_with("healthy,")),
+        "{stdout}"
+    );
+    assert!(
+        stdout.lines().any(|line| line.starts_with("peer,")),
+        "{stdout}"
+    );
+    assert!(
+        stderr.contains("peer closure (1 target session)"),
+        "{stderr}"
+    );
+    assert!(
+        !stderr.contains("managed client group was cancelled"),
+        "{stderr}"
+    );
+}
+
+#[test]
+fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() {
+    let params = test_params(None, Duration::from_millis(10));
+    let (failure_tx, failure_rx) = mpsc::channel();
+    let (healthy_reply_tx, healthy_reply_rx) = mpsc::channel();
+    let failure = start_open_failure_server_with_signal(params.clone(), failure_tx);
+    let healthy = start_echo_server_with_first_reply(params.clone(), healthy_reply_tx);
+    let peer_closed = start_gated_peer_close_server(params, vec![failure_rx, healthy_reply_rx]);
+    let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
+    command.args([
+        "--duration",
+        "0s",
+        "--interval",
+        "10ms",
+        "--format",
+        "csv",
+        "--columns",
+        "target,seq",
+        "--header",
+        "never",
         "--target",
         &format!("failure={}", failure.addr),
+        "--target",
+        &format!("healthy={}", healthy.addr),
+        "--target",
+        &format!("peer={}", peer_closed.addr),
     ]);
 
     let output = run_with_timeout(command, Duration::from_secs(3));
-    peer_closed.join();
     failure.join();
+    healthy.join();
+    peer_closed.join();
 
     assert!(
         !output.status.success(),
@@ -375,6 +460,10 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() {
         stdout.lines().any(|line| line.starts_with("peer,")),
         "{stdout}"
     );
+    assert!(
+        stdout.lines().any(|line| line.starts_with("healthy,")),
+        "{stdout}"
+    );
     assert!(stderr.contains("target failure failed"), "{stderr}");
     assert!(stderr.contains("zero token without close flag"), "{stderr}");
     assert!(
@@ -383,6 +472,67 @@ fn continuous_mixed_peer_close_and_open_failure_exits_nonzero() {
     );
 }
 
+#[cfg(unix)]
+#[test]
+fn explicit_group_interruption_succeeds_without_peer_close_error() {
+    let params = test_params(None, Duration::from_millis(10));
+    let a = start_interruptible_echo_server(params.clone());
+    let b = start_interruptible_echo_server(params);
+    let mut command = Command::new(env!("CARGO_BIN_EXE_irtt-cli"));
+    command.args([
+        "--duration",
+        "0s",
+        "--interval",
+        "10ms",
+        "--format",
+        "csv",
+        "--columns",
+        "target,seq",
+        "--header",
+        "never",
+        "--target",
+        &format!("a={}", a.addr),
+        "--target",
+        &format!("b={}", b.addr),
+    ]);
+
+    let child = command
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .unwrap();
+    a.wait_until_opened();
+    b.wait_until_opened();
+    thread::sleep(Duration::from_millis(50));
+    let signal_status = Command::new("kill")
+        .args(["-INT", &child.id().to_string()])
+        .status()
+        .unwrap();
+    assert!(signal_status.success());
+    let output = wait_with_timeout(child, Duration::from_secs(2));
+    a.join();
+    b.join();
+
+    assert!(
+        output.status.success(),
+        "status={:?}\nstderr={}",
+        output.status.code(),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let stdout = String::from_utf8(output.stdout).unwrap();
+    let stderr = String::from_utf8(output.stderr).unwrap();
+    assert!(
+        stdout.lines().any(|line| line.starts_with("a,")),
+        "{stdout}"
+    );
+    assert!(
+        stdout.lines().any(|line| line.starts_with("b,")),
+        "{stdout}"
+    );
+    assert!(stderr.contains("interrupted, closing group"), "{stderr}");
+    assert!(!stderr.contains("peer closure"), "{stderr}");
+}
+
 #[test]
 fn all_open_failures_exit_nonzero_with_diagnostics() {
     let params = test_params(None, Duration::from_millis(10));
@@ -420,11 +570,15 @@ fn all_open_failures_exit_nonzero_with_diagnostics() {
 }
 
 fn run_with_timeout(mut command: Command, timeout: Duration) -> Output {
-    let mut child = command
+    let child = command
         .stdout(Stdio::piped())
         .stderr(Stdio::piped())
         .spawn()
         .unwrap();
+    wait_with_timeout(child, timeout)
+}
+
+fn wait_with_timeout(mut child: std::process::Child, timeout: Duration) -> Output {
     let deadline = Instant::now() + timeout;
     loop {
         if child.try_wait().unwrap().is_some() {
@@ -452,6 +606,20 @@ fn run_with_timeout(mut command: Command, timeout: Duration) -> Output {
 }
 
 fn start_echo_server(params: Params) -> FakeServer {
+    start_echo_server_inner(params, None)
+}
+
+fn start_echo_server_with_first_reply(
+    params: Params,
+    first_reply_tx: mpsc::Sender<()>,
+) -> FakeServer {
+    start_echo_server_inner(params, Some(first_reply_tx))
+}
+
+fn start_echo_server_inner(
+    params: Params,
+    mut first_reply_tx: Option<mpsc::Sender<()>>,
+) -> FakeServer {
     let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
     socket
         .set_read_timeout(Some(Duration::from_secs(2)))
@@ -482,12 +650,62 @@ fn start_echo_server(params: Params) -> FakeServer {
                     peer,
                 )
                 .unwrap();
+            if let Some(tx) = first_reply_tx.take() {
+                tx.send(()).unwrap();
+            }
         }
     });
     FakeServer { addr, done }
 }
 
+fn start_interruptible_echo_server(params: Params) -> InterruptibleFakeServer {
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(2)))
+        .unwrap();
+    let addr = socket.local_addr().unwrap();
+    let (opened_tx, opened) = mpsc::channel();
+    let done = thread::spawn(move || {
+        let (_, peer) = recv_request(&socket);
+        opened_tx.send(()).unwrap();
+        socket
+            .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
+            .unwrap();
+        socket
+            .set_read_timeout(Some(Duration::from_millis(500)))
+            .unwrap();
+
+        while let Some((packet, peer)) = recv_request_timeout(&socket) {
+            let flags = packet[3];
+            if flags & FLAG_CLOSE != 0 {
+                break;
+            }
+            let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
+            socket
+                .send_to(
+                    &echo_reply_packet(
+                        TOKEN,
+                        seq,
+                        &params,
+                        &TimestampFields::default(),
+                        FLAG_REPLY,
+                    ),
+                    peer,
+                )
+                .unwrap();
+        }
+    });
+    InterruptibleFakeServer { addr, opened, done }
+}
+
 fn start_peer_close_server(params: Params) -> FakeServer {
+    start_gated_peer_close_server(params, Vec::new())
+}
+
+fn start_gated_peer_close_server(
+    params: Params,
+    release_gates: Vec<mpsc::Receiver<()>>,
+) -> FakeServer {
     let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
     socket
         .set_read_timeout(Some(Duration::from_secs(2)))
@@ -499,6 +717,10 @@ fn start_peer_close_server(params: Params) -> FakeServer {
             .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, TOKEN, &params), peer)
             .unwrap();
         let (packet, peer) = recv_request(&socket);
+        for gate in release_gates {
+            gate.recv_timeout(Duration::from_secs(2))
+                .expect("timed out waiting to release peer-close reply");
+        }
         let seq = u32::from_le_bytes(packet[12..16].try_into().unwrap());
         socket
             .send_to(
@@ -517,6 +739,20 @@ fn start_peer_close_server(params: Params) -> FakeServer {
 }
 
 fn start_open_failure_server(params: Params) -> FakeServer {
+    start_open_failure_server_inner(params, None)
+}
+
+fn start_open_failure_server_with_signal(
+    params: Params,
+    failure_tx: mpsc::Sender<()>,
+) -> FakeServer {
+    start_open_failure_server_inner(params, Some(failure_tx))
+}
+
+fn start_open_failure_server_inner(
+    params: Params,
+    failure_tx: Option<mpsc::Sender<()>>,
+) -> FakeServer {
     let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
     socket
         .set_read_timeout(Some(Duration::from_secs(2)))
@@ -527,6 +763,9 @@ fn start_open_failure_server(params: Params) -> FakeServer {
         socket
             .send_to(&open_reply(FLAG_OPEN | FLAG_REPLY, 0, &params), peer)
             .unwrap();
+        if let Some(tx) = failure_tx {
+            tx.send(()).unwrap();
+        }
     });
     FakeServer { addr, done }
 }
@@ -537,6 +776,14 @@ fn recv_request(socket: &UdpSocket) -> (Vec<u8>, SocketAddr) {
     (buf[..size].to_vec(), peer)
 }
 
+fn recv_request_timeout(socket: &UdpSocket) -> Option<(Vec<u8>, SocketAddr)> {
+    let mut buf = [0_u8; 2048];
+    socket
+        .recv_from(&mut buf)
+        .ok()
+        .map(|(size, peer)| (buf[..size].to_vec(), peer))
+}
+
 fn open_reply(flags: u8, token: u64, params: &Params) -> Vec<u8> {
     let mut packet = Vec::new();
     packet.extend_from_slice(&MAGIC);