fix(cli): drive single-target runs with ManagedClient

Replace the manual single-target ClientSession loops in the stream and TUI adapters with ManagedClient::start_with_subscription so the existing worker owns opening, deadline-aware pacing, receive handling, timeout polling, draining, and close.

Use bounded subscription waits for interruption and input polling, drain queued events before and after join, and report per-subscription drops without changing output, statistics, labels, summaries, warning, or peer-close policy. Add process and unit regressions for 10 ms cadence, interruption, final-event draining, TUI terminal state, and overflow visibility.

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

@@ -9,9 +9,10 @@ use std::{
 };
 
 use irtt_client::{
-    ClientEvent, EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig,
-    ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, ManagedTargetOutcome,
-    SubscriberConfig, SubscriberOverflow, TargetEvent,
+    ClientEvent, EventSubscriptionError, ManagedClient, ManagedClientGroup,
+    ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason,
+    ManagedGroupEvent, ManagedTargetOutcome, SessionEndReason, SubscriberConfig,
+    SubscriberOverflow, TargetEvent,
 };
 #[cfg(all(test, feature = "stats"))]
 use irtt_client::{ClientTimestamp, PacketMeta, RttSample, SignedDuration};
@@ -28,7 +29,6 @@ use crate::shared::client::{
         peer_close_run_error, request_group_stop_once, should_print_final_summary,
         should_stop_group_on_peer_close,
     },
-    ClientSession,
 };
 
 #[cfg(feature = "stats")]
@@ -49,6 +49,8 @@ const FINITE_STATS_MEMORY_VERY_STRONG_WARNING_BYTES: u64 = GIB;
 
 const GROUP_IDLE_SLEEP: Duration = Duration::from_millis(5);
 const GROUP_COMPLETION_GRACE: Duration = Duration::from_secs(1);
+const MANAGED_EVENT_WAIT_SLICE: Duration = Duration::from_millis(20);
+const MANAGED_EVENT_CAPACITY: usize = 16_384;
 
 pub fn run_stream(
     args: ClientArgs,
@@ -103,66 +105,71 @@ pub fn run_stream(
         return Ok(());
     }
 
-    let mut session = ClientSession::connect(args.to_client_config(), continuous)?;
-
-    if is_shutdown_requested(shutdown_requested) {
-        return Ok(());
-    }
-
-    let events = session.open()?;
-    #[cfg(feature = "stats")]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?;
-    #[cfg(not(feature = "stats"))]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label))?;
-
+    let (session, events) = ManagedClient::start_with_subscription(
+        args.to_client_config(),
+        managed_event_subscriber_config(),
+    )?;
     let mut interrupted = false;
-    while session.should_continue(shutdown_requested) {
-        let events = session.step(shutdown_requested)?;
-        #[cfg(feature = "stats")]
-        print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?;
-        #[cfg(not(feature = "stats"))]
-        print_events_with_stats(&mut stream_output, &events, Some(target_label))?;
-
+    loop {
         if is_shutdown_requested(shutdown_requested) {
             interrupted = true;
+            session.stop();
             break;
         }
 
-        session.sleep_until_next_send();
+        match events.recv_timeout(MANAGED_EVENT_WAIT_SLICE) {
+            Ok(Some(event)) => {
+                #[cfg(feature = "stats")]
+                print_events_with_stats(
+                    &mut stream_output,
+                    std::slice::from_ref(&event),
+                    Some(target_label),
+                    &mut stats,
+                )?;
+                #[cfg(not(feature = "stats"))]
+                print_events_with_stats(
+                    &mut stream_output,
+                    std::slice::from_ref(&event),
+                    Some(target_label),
+                )?;
+            }
+            Ok(None) => {}
+            Err(EventSubscriptionError::Disconnected) => break,
+        }
     }
     interrupted |= is_shutdown_requested(shutdown_requested);
 
     if interrupted {
         eprintln!("interrupted, closing session...");
-    }
-
-    if session.should_drain_final(interrupted) {
-        session.drain_final(|events| {
+        drain_single_client_events(
+            &events,
+            &mut stream_output,
+            target_label,
             #[cfg(feature = "stats")]
-            let _ =
-                print_events_with_stats(&mut stream_output, events, Some(target_label), &mut stats);
-            #[cfg(not(feature = "stats"))]
-            let _ = print_events_with_stats(&mut stream_output, events, Some(target_label));
-        })?;
+            &mut stats,
+        )?;
     }
 
-    let events = session.poll_timeouts()?;
-    #[cfg(feature = "stats")]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?;
-    #[cfg(not(feature = "stats"))]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label))?;
+    let outcome = session.join();
+    drain_single_client_events(
+        &events,
+        &mut stream_output,
+        target_label,
+        #[cfg(feature = "stats")]
+        &mut stats,
+    )?;
+    if let Some(warning) = dropped_event_warning(events.dropped_events(), "managed client") {
+        eprintln!("{warning}");
+    }
 
-    let events = session.close()?;
-    #[cfg(feature = "stats")]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label), &mut stats)?;
-    #[cfg(not(feature = "stats"))]
-    print_events_with_stats(&mut stream_output, &events, Some(target_label))?;
     interrupted |= is_shutdown_requested(shutdown_requested);
-    let peer_close_error = peer_close_run_error(
-        continuous,
-        interrupted,
-        if session.is_peer_closed() { 1 } else { 0 },
-    );
+    let peer_close_error = outcome.as_ref().ok().and_then(|outcome| {
+        peer_close_run_error(
+            continuous,
+            interrupted,
+            u64::from(outcome.end_reason == SessionEndReason::PeerClosed),
+        )
+    });
     let print_final_summary = should_print_final_summary(continuous, interrupted);
     stream_output.print_final_summary = print_final_summary;
     stream_output.show_running_only_summary_note = continuous && interrupted && print_final_summary;
@@ -171,12 +178,59 @@ pub fn run_stream(
     #[cfg(not(feature = "stats"))]
     stream_output.print_summary()?;
     stream_output.out.flush()?;
+    if let Err(error) = outcome {
+        return Err(error.into());
+    }
     if let Some(error) = peer_close_error {
         return Err(error.into());
     }
     Ok(())
 }
 
+fn managed_event_subscriber_config() -> SubscriberConfig {
+    SubscriberConfig {
+        capacity: MANAGED_EVENT_CAPACITY,
+        // CLI output is best-effort under sustained backpressure. Dropping
+        // stale rows keeps the managed worker independent from output speed,
+        // while dropped_events() makes incomplete output visible.
+        overflow: SubscriberOverflow::DropOldest,
+    }
+}
+
+#[cfg(feature = "stats")]
+fn drain_single_client_events<W: Write>(
+    events: &irtt_client::EventSubscription,
+    stream_output: &mut StreamOutput<'_, W>,
+    target_label: &str,
+    stats: &mut StatsCollector,
+) -> io::Result<()> {
+    while let Ok(Some(event)) = events.try_recv() {
+        print_events_with_stats(
+            stream_output,
+            std::slice::from_ref(&event),
+            Some(target_label),
+            stats,
+        )?;
+    }
+    Ok(())
+}
+
+#[cfg(not(feature = "stats"))]
+fn drain_single_client_events<W: Write>(
+    events: &irtt_client::EventSubscription,
+    stream_output: &mut StreamOutput<'_, W>,
+    target_label: &str,
+) -> io::Result<()> {
+    while let Ok(Some(event)) = events.try_recv() {
+        print_events_with_stats(
+            stream_output,
+            std::slice::from_ref(&event),
+            Some(target_label),
+        )?;
+    }
+    Ok(())
+}
+
 struct StreamOutput<'a, W: Write> {
     config: OutputConfig,
     header_printed: bool,
@@ -300,7 +354,7 @@ fn run_group_stream(
         group_config,
         managed_targets,
         SubscriberConfig {
-            capacity: 16_384,
+            capacity: MANAGED_EVENT_CAPACITY,
             // CLI output is best-effort under sustained backpressure. Dropping
             // stale rows keeps continuous runs attached to the running group
             // instead of turning a full output queue into a disconnected
@@ -427,7 +481,7 @@ fn run_group_stream(
             }
         }
     }
-    if let Some(warning) = dropped_event_warning(events.dropped_events()) {
+    if let Some(warning) = dropped_event_warning(events.dropped_events(), "managed group") {
         eprintln!("{warning}");
     }
 
@@ -492,10 +546,15 @@ fn report_target_failure(target: &ManagedTargetOutcome) {
     }
 }
 
-fn dropped_event_warning(dropped_events: u64) -> Option<String> {
+fn dropped_event_warning(dropped_events: u64, source: &str) -> Option<String> {
     (dropped_events > 0).then(|| {
+        let event_word = if dropped_events == 1 {
+            "event"
+        } else {
+            "events"
+        };
         format!(
-            "irtt-rs: warning: dropped {dropped_events} managed group events; output and statistics may be incomplete"
+            "irtt-rs: warning: dropped {dropped_events} {source} {event_word}; output and statistics may be incomplete"
         )
     })
 }
@@ -654,12 +713,57 @@ mod tests {
 
     #[test]
     fn dropped_event_warning_marks_statistics_incomplete() {
-        assert_eq!(dropped_event_warning(0), None);
-        let warning = dropped_event_warning(7).unwrap();
+        assert_eq!(dropped_event_warning(0, "managed client"), None);
+        let warning = dropped_event_warning(7, "managed group").unwrap();
         assert!(warning.contains("dropped 7 managed group events"));
         assert!(warning.contains("statistics may be incomplete"));
     }
 
+    #[test]
+    fn single_client_drain_preserves_final_queue_and_surfaces_overflow() {
+        let hub = irtt_client::EventHub::new();
+        let events = hub
+            .subscribe(SubscriberConfig {
+                capacity: 1,
+                overflow: SubscriberOverflow::DropOldest,
+            })
+            .unwrap();
+        hub.publish(reply_event(1, 1_000));
+        hub.publish(reply_event(2, 2_000));
+        hub.disconnect_all();
+
+        let mut stats = StatsCollector::new(StatsConfig::finite());
+        let mut out = Vec::new();
+        {
+            let mut stream_output = StreamOutput {
+                config: output_config(
+                    crate::cmd::client::OutputFormat::Csv,
+                    Some("event,seq"),
+                    crate::cmd::client::HeaderMode::Never,
+                    false,
+                ),
+                header_printed: false,
+                print_final_summary: false,
+                show_running_only_summary_note: false,
+                out: &mut out,
+            };
+            drain_single_client_events(&events, &mut stream_output, "target", &mut stats).unwrap();
+        }
+
+        let rendered = String::from_utf8(out).unwrap();
+        assert_eq!(rendered.trim(), "echo_reply,2");
+        assert_eq!(events.dropped_events(), 1);
+        assert!(
+            dropped_event_warning(events.dropped_events(), "managed client")
+                .unwrap()
+                .contains("dropped 1 managed client event")
+        );
+        assert_eq!(
+            events.try_recv().unwrap_err(),
+            EventSubscriptionError::Disconnected
+        );
+    }
+
     #[test]
     fn output_helper_streams_and_collects_events() {
         let sent_at = test_timestamp(Duration::from_secs(1));

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

@@ -7,9 +7,9 @@ use std::{
 
 use crossterm::event::{self, Event, KeyCode, KeyEventKind, KeyModifiers};
 use irtt_client::{
-    EventSubscriptionError, ManagedClientGroup, ManagedClientGroupConfig,
-    ManagedGroupCompletionPolicy, ManagedGroupEndReason, ManagedGroupEvent, SubscriberConfig,
-    SubscriberOverflow,
+    EventSubscription, EventSubscriptionError, ManagedClient, ManagedClientGroup,
+    ManagedClientGroupConfig, ManagedGroupCompletionPolicy, ManagedGroupEndReason,
+    ManagedGroupEvent, SessionEndReason, SubscriberConfig, SubscriberOverflow,
 };
 
 use crate::{
@@ -17,7 +17,6 @@ use crate::{
     shared::client::{
         is_shutdown_requested,
         session::{peer_close_run_error, request_group_stop_once, should_stop_group_on_peer_close},
-        ClientSession,
     },
 };
 
@@ -27,6 +26,7 @@ const RENDER_INTERVAL: Duration = Duration::from_millis(250);
 const TUI_WAIT_SLICE: Duration = Duration::from_millis(20);
 const IDLE_SLEEP: Duration = Duration::from_millis(5);
 const GROUP_COMPLETION_GRACE: Duration = Duration::from_secs(1);
+const MANAGED_EVENT_CAPACITY: usize = 16_384;
 
 pub fn run_tui(
     args: TuiArgs,
@@ -61,7 +61,10 @@ pub fn run_tui(
         );
     }
 
-    let mut session = match ClientSession::connect(args.to_client_config(), continuous) {
+    let (session, events) = match ManagedClient::start_with_subscription(
+        args.to_client_config(),
+        managed_event_subscriber_config(),
+    ) {
         Ok(session) => session,
         Err(err) => {
             state.set_error(err.to_string());
@@ -70,43 +73,33 @@ pub fn run_tui(
         }
     };
 
-    if is_shutdown_requested(shutdown_requested) {
-        return Ok(());
-    }
-
-    let events = session.open()?;
-    state.process_events(&events);
-    render_if_due(&mut terminal, &state, &mut next_render, true)?;
-
     let mut interrupted = false;
-    // Keep this in lockstep with run_stream: send due probes, drain available
-    // replies, poll timeouts, sleep toward the next absolute send deadline,
-    // then perform the same final drain, timeout poll, and close sequence.
-    while session.should_continue(shutdown_requested) {
+    loop {
+        if is_shutdown_requested(shutdown_requested) {
+            interrupted = true;
+            session.stop();
+            break;
+        }
+
         if handle_input(&mut state, shutdown_requested)? {
             render_if_due(&mut terminal, &state, &mut next_render, true)?;
         }
         if state.quit_requested {
             interrupted = true;
+            session.stop();
             break;
         }
 
-        let events = session.step(shutdown_requested)?;
-        state.process_events(&events);
-
-        if is_shutdown_requested(shutdown_requested) {
-            interrupted = true;
-            break;
+        match events.recv_timeout(managed_tui_wait_duration(&next_render, state.paused)) {
+            Ok(Some(event)) => {
+                state.process_event(&event);
+                render_if_due(&mut terminal, &state, &mut next_render, false)?;
+            }
+            Ok(None) => {
+                render_if_due(&mut terminal, &state, &mut next_render, false)?;
+            }
+            Err(EventSubscriptionError::Disconnected) => break,
         }
-
-        render_if_due(&mut terminal, &state, &mut next_render, false)?;
-        wait_for_tui_activity(
-            session.next_send_deadline(),
-            &mut next_render,
-            &mut state,
-            &mut terminal,
-            shutdown_requested,
-        )?;
     }
     interrupted |= is_shutdown_requested(shutdown_requested);
 
@@ -115,29 +108,32 @@ pub fn run_tui(
         render_if_due(&mut terminal, &state, &mut next_render, true)?;
     }
 
-    if session.should_drain_final(interrupted) {
+    if interrupted {
         state.set_status(TuiStatus::Draining);
-        let mut drain_render = Instant::now();
-        session.drain_final(|events| {
-            state.process_events(events);
-            let _ = render_if_due(&mut terminal, &state, &mut drain_render, false);
-        })?;
+        drain_single_tui_events(&events, &mut state);
         render_if_due(&mut terminal, &state, &mut next_render, true)?;
     }
 
-    let events = session.poll_timeouts()?;
-    state.process_events(&events);
-
     state.set_status(TuiStatus::Closing);
     render_if_due(&mut terminal, &state, &mut next_render, true)?;
 
-    let events = session.close()?;
-    state.process_events(&events);
+    let outcome = session.join();
+    drain_single_tui_events(&events, &mut state);
+    state.mark_dropped_client_events(events.dropped_events());
+    let outcome = match outcome {
+        Ok(outcome) => outcome,
+        Err(err) => {
+            state.set_error(err.to_string());
+            render_if_due(&mut terminal, &state, &mut next_render, true)?;
+            return Err(Box::new(err));
+        }
+    };
+
     interrupted |= is_shutdown_requested(shutdown_requested);
     if let Some(error) = peer_close_run_error(
         continuous,
         interrupted,
-        if session.is_peer_closed() { 1 } else { 0 },
+        u64::from(outcome.end_reason == SessionEndReason::PeerClosed),
     ) {
         state.set_run_error(error.clone());
         render_if_due(&mut terminal, &state, &mut next_render, true)?;
@@ -175,7 +171,7 @@ fn run_group_tui(
         group_config,
         managed_targets,
         SubscriberConfig {
-            capacity: 16_384,
+            capacity: MANAGED_EVENT_CAPACITY,
             overflow: SubscriberOverflow::DropOldest,
         },
     )?;
@@ -272,7 +268,7 @@ fn run_group_tui(
             }
         }
     }
-    state.mark_dropped_events(events.dropped_events());
+    state.mark_dropped_group_events(events.dropped_events());
     for target in &outcome.targets {
         if terminal_targets.insert(target.id.as_str().to_owned()) {
             state.process_target_outcome(target);
@@ -451,6 +447,28 @@ fn render_if_due(
     Ok(())
 }
 
+fn managed_event_subscriber_config() -> SubscriberConfig {
+    SubscriberConfig {
+        capacity: MANAGED_EVENT_CAPACITY,
+        overflow: SubscriberOverflow::DropOldest,
+    }
+}
+
+fn drain_single_tui_events(events: &EventSubscription, state: &mut TuiState) {
+    while let Ok(Some(event)) = events.try_recv() {
+        state.process_event(&event);
+    }
+}
+
+fn managed_tui_wait_duration(next_render: &Instant, paused: bool) -> Duration {
+    let render_wait = if paused {
+        TUI_WAIT_SLICE
+    } else {
+        next_render.saturating_duration_since(Instant::now())
+    };
+    render_wait.min(TUI_WAIT_SLICE)
+}
+
 fn wait_for_tui_activity(
     next_send_deadline: Option<Instant>,
     next_render: &mut Instant,
@@ -552,4 +570,30 @@ mod tests {
             &irtt_client::ManagedTargetEndReason::PeerClosed
         ));
     }
+
+    #[test]
+    fn single_target_managed_drain_consumes_queued_terminal_event() {
+        let hub = irtt_client::EventHub::new();
+        let events = hub
+            .subscribe(SubscriberConfig {
+                capacity: 4,
+                overflow: SubscriberOverflow::DropOldest,
+            })
+            .unwrap();
+        hub.publish(irtt_client::ClientEvent::SessionClosed {
+            remote: "127.0.0.1:2112".parse().unwrap(),
+            token: 7,
+            at: irtt_client::ClientTimestamp::now(),
+        });
+        hub.disconnect_all();
+        let mut state = TuiState::new(TuiConfig::default());
+
+        drain_single_tui_events(&events, &mut state);
+
+        assert_eq!(state.status(), TuiStatus::Complete);
+        assert_eq!(
+            events.try_recv().unwrap_err(),
+            EventSubscriptionError::Disconnected
+        );
+    }
 }

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

@@ -140,12 +140,6 @@ impl TuiState {
         }
     }
 
-    pub(super) fn process_events(&mut self, events: &[ClientEvent]) {
-        for event in events {
-            self.process_event(event);
-        }
-    }
-
     pub(super) fn process_event(&mut self, event: &ClientEvent) {
         self.process_event_for_target(0, event);
     }
@@ -388,13 +382,31 @@ impl TuiState {
         self.status = status;
     }
 
-    pub(super) fn mark_dropped_events(&mut self, dropped_events: u64) {
+    #[cfg(test)]
+    pub(super) fn status(&self) -> TuiStatus {
+        self.status
+    }
+
+    pub(super) fn mark_dropped_client_events(&mut self, dropped_events: u64) {
+        self.mark_dropped_events(dropped_events, "managed client");
+    }
+
+    pub(super) fn mark_dropped_group_events(&mut self, dropped_events: u64) {
+        self.mark_dropped_events(dropped_events, "managed group");
+    }
+
+    fn mark_dropped_events(&mut self, dropped_events: u64, source: &str) {
         if dropped_events == 0 {
             return;
         }
         self.dropped_events = dropped_events;
+        let event_word = if dropped_events == 1 {
+            "event"
+        } else {
+            "events"
+        };
         let warning =
-            format!("dropped {dropped_events} managed group events; statistics may be incomplete");
+            format!("dropped {dropped_events} {source} {event_word}; statistics may be incomplete");
         self.last_warning = Some(warning.clone());
         self.push_event(format!("warning {warning}"));
     }
@@ -2497,7 +2509,7 @@ mod tests {
     fn dropped_events_mark_tui_statistics_incomplete() {
         let mut state = TuiState::default();
 
-        state.mark_dropped_events(9);
+        state.mark_dropped_group_events(9);
 
         assert_eq!(state.dropped_events, 9);
         assert!(state
@@ -2510,6 +2522,19 @@ mod tests {
             .is_some_and(|event| event.contains("dropped 9 managed group events")));
     }
 
+    #[test]
+    fn dropped_single_client_events_are_visible() {
+        let mut state = TuiState::default();
+
+        state.mark_dropped_client_events(3);
+
+        assert_eq!(state.dropped_events, 3);
+        assert!(state
+            .recent_events
+            .back()
+            .is_some_and(|event| event.contains("dropped 3 managed client events")));
+    }
+
     #[test]
     fn recent_event_buffer_stays_bounded() {
         let mut state = TuiState::default();