tests: rework the actually useful tests to live in the integration crate

crates/irtt-integration-tests/tests/async_client.rs

@@ -75,16 +75,48 @@ async fn exercise_client_server() {
             ..
         }] if *remote == server_addr
     ));
-    assert!(matches!(
-        client.recv().await.unwrap().as_slice(),
-        [ClientEvent::EchoReply {
-            seq: 0,
-            remote,
-            server_timing: Some(_),
-            received_stats: Some(_),
-            ..
-        }] if *remote == server_addr
-    ));
+    let events = client.recv().await.unwrap();
+    let [ClientEvent::EchoReply {
+        seq: 0,
+        remote,
+        sent_at,
+        received_at,
+        rtt,
+        server_timing: Some(timing),
+        one_way: Some(one_way),
+        received_stats: Some(_),
+        packet_meta,
+        ..
+    }] = events.as_slice()
+    else {
+        panic!("expected a measured reply, got {events:?}");
+    };
+    assert_eq!(*remote, server_addr);
+    assert_eq!(rtt.raw, received_at.mono.duration_since(sent_at.mono));
+    // Verify the measurement endpoint through public reply metadata and real
+    // OS timestamps. This also covers the userspace fallback on other targets.
+    let receive_wall = packet_meta
+        .kernel_rx_timestamp
+        .filter(|kernel| {
+            received_at
+                .wall
+                .duration_since(*kernel)
+                .is_ok_and(|lag| lag <= Duration::from_secs(1))
+        })
+        .unwrap_or(received_at.wall);
+    let receive_ns = i128::try_from(
+        receive_wall
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos(),
+    )
+    .unwrap();
+    assert_eq!(
+        one_way.server_to_client,
+        Some(irtt_client::SignedDuration::from_nanos(
+            receive_ns - i128::from(timing.send_wall_ns.unwrap())
+        ))
+    );
     assert!(matches!(
         client.close().await.unwrap().as_slice(),
         [ClientEvent::SessionClosed { remote, .. }] if *remote == server_addr

crates/irtt-integration-tests/tests/async_client_lifecycle.rs

@@ -0,0 +1,678 @@
+use std::{
+    future::Future,
+    net::UdpSocket,
+    pin::Pin,
+    sync::mpsc,
+    task::{Context, Poll, Waker},
+    thread::{self, JoinHandle},
+};
+
+use irtt_proto::{
+    decode_request, encode_echo_reply, encode_open_reply, flags, verify_packet_hmac,
+    DecodedRequestKind, EchoReply, OpenReply, Params, ReceivedStats, StampAt, TimestampFields,
+};
+use tokio::runtime::{Builder, Runtime};
+
+mod support;
+
+use irtt_client::{AsyncClient, ClientConfig, ClientError, ClientEvent, OpenOutcome};
+use irtt_client::{Client, RunMode, SocketConfig};
+use irtt_server::ServerConfig;
+use std::{
+    io,
+    net::SocketAddr,
+    time::{Duration, Instant},
+};
+use support::InTreeServer;
+
+const TOKEN: u64 = 0x1234_5678_90ab_cdef;
+
+struct TestServer {
+    addr: SocketAddr,
+    packets: mpsc::Receiver<Vec<u8>>,
+    done: JoinHandle<()>,
+}
+
+impl TestServer {
+    fn finish(self) -> Vec<Vec<u8>> {
+        self.done.join().unwrap();
+        self.packets.try_iter().collect()
+    }
+}
+
+fn runtime() -> Runtime {
+    Builder::new_current_thread()
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap()
+}
+
+fn config(addr: SocketAddr, key: Option<Vec<u8>>, dscp: u8) -> ClientConfig {
+    ClientConfig {
+        server_addr: addr.to_string(),
+        received_stats: ReceivedStats::None,
+        stamp_at: StampAt::None,
+        dscp,
+        hmac_key: key,
+        open_timeouts: vec![Duration::from_millis(200)],
+        socket_config: SocketConfig {
+            recv_timeout: Some(Duration::from_millis(200)),
+            ..SocketConfig::default()
+        },
+        ..ClientConfig::default()
+    }
+}
+
+async fn opened_client(addr: SocketAddr, dscp: u8) -> AsyncClient {
+    let mut client = AsyncClient::connect(config(addr, None, dscp))
+        .await
+        .unwrap();
+    client.open().await.unwrap();
+    client
+}
+
+fn start_server<F>(handler: F) -> TestServer
+where
+    F: FnOnce(UdpSocket, mpsc::Sender<Vec<u8>>) + Send + 'static,
+{
+    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 (tx, packets) = mpsc::channel();
+    let done = thread::spawn(move || handler(socket, tx));
+    TestServer {
+        addr,
+        packets,
+        done,
+    }
+}
+
+fn recv_packet(socket: &UdpSocket, tx: &mpsc::Sender<Vec<u8>>) -> (Vec<u8>, SocketAddr) {
+    let mut buffer = [0_u8; 8192];
+    let (len, peer) = socket.recv_from(&mut buffer).unwrap();
+    let packet = buffer[..len].to_vec();
+    tx.send(packet.clone()).unwrap();
+    (packet, peer)
+}
+
+/// Admits a request the way a compliant server does: structural decode first,
+/// then HMAC presence policy, then authentication. Returns `None` for anything
+/// a server would silently discard.
+fn decode_server_request<'a>(
+    packet: &'a [u8],
+    key: Option<&[u8]>,
+) -> Option<DecodedRequestKind<'a>> {
+    let request = decode_request(packet).ok()?;
+    if request.hmac_present != key.is_some() {
+        return None;
+    }
+    if let Some(key) = key {
+        verify_packet_hmac(key, packet).ok()?;
+    }
+    Some(request.kind)
+}
+
+/// Returns the requested parameters and whether this was a no-test open.
+fn open_request(packet: &[u8], key: Option<&[u8]>) -> (Params, bool) {
+    match decode_server_request(packet, key) {
+        Some(DecodedRequestKind::Open { params, no_test }) => {
+            (Params::decode(params).unwrap(), no_test)
+        }
+        other => panic!("expected an authenticated open request, got {other:?}"),
+    }
+}
+
+fn echo_request_sequence(packet: &[u8], key: Option<&[u8]>) -> u32 {
+    match decode_server_request(packet, key) {
+        Some(DecodedRequestKind::Echo { sequence, .. }) => sequence,
+        other => panic!("expected an authenticated echo request, got {other:?}"),
+    }
+}
+
+fn close_request_token(packet: &[u8], key: Option<&[u8]>) -> u64 {
+    match decode_server_request(packet, key) {
+        Some(DecodedRequestKind::Close { token }) => token,
+        other => panic!("expected an authenticated close request, got {other:?}"),
+    }
+}
+
+fn send_open_reply(
+    socket: &UdpSocket,
+    peer: SocketAddr,
+    params: Params,
+    key: Option<&[u8]>,
+    reply_flags: u8,
+    token: u64,
+) {
+    let reply = encode_open_reply(
+        &OpenReply {
+            flags: reply_flags,
+            token,
+            params,
+        },
+        key,
+    )
+    .unwrap();
+    socket.send_to(&reply, peer).unwrap();
+}
+
+fn open_session(
+    socket: &UdpSocket,
+    tx: &mpsc::Sender<Vec<u8>>,
+    key: Option<&[u8]>,
+) -> (Params, SocketAddr) {
+    let (packet, peer) = recv_packet(socket, tx);
+    let (params, _) = open_request(&packet, key);
+    send_open_reply(
+        socket,
+        peer,
+        params.clone(),
+        key,
+        flags::FLAG_OPEN | flags::FLAG_REPLY,
+        TOKEN,
+    );
+    (params, peer)
+}
+
+fn echo_reply(
+    params: &Params,
+    sequence: u32,
+    token: u64,
+    reply_flags: u8,
+    key: Option<&[u8]>,
+) -> Vec<u8> {
+    encode_echo_reply(
+        &EchoReply {
+            flags: reply_flags,
+            token,
+            sequence,
+            recv_count: None,
+            recv_window: None,
+            timestamps: TimestampFields::default(),
+            payload: Vec::new(),
+        },
+        params,
+        key,
+    )
+    .unwrap()
+}
+
+fn start_no_test_server() -> TestServer {
+    start_server(move |socket, tx| {
+        let (open_packet, peer) = recv_packet(&socket, &tx);
+        let (params, no_test) = open_request(&open_packet, None);
+        assert!(no_test);
+        send_open_reply(
+            &socket,
+            peer,
+            params,
+            None,
+            flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE,
+            0,
+        );
+    })
+}
+
+fn start_peer_close_server() -> TestServer {
+    start_server(move |socket, tx| {
+        let (params, peer) = open_session(&socket, &tx, None);
+        let (probe_packet, _) = recv_packet(&socket, &tx);
+        let sequence = echo_request_sequence(&probe_packet, None);
+        socket
+            .send_to(
+                &echo_reply(
+                    &params,
+                    sequence,
+                    TOKEN,
+                    flags::FLAG_REPLY | flags::FLAG_CLOSE,
+                    None,
+                ),
+                peer,
+            )
+            .unwrap();
+    })
+}
+
+fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
+    let mut context = Context::from_waker(Waker::noop());
+    Future::poll(future, &mut context)
+}
+
+fn poll_recv_once(client: &mut AsyncClient) -> Poll<Result<Vec<ClientEvent>, ClientError>> {
+    let mut future = Box::pin(client.recv());
+    poll_once(Pin::as_mut(&mut future))
+}
+
+#[test]
+fn connect_requires_current_runtime_when_polled() {
+    let mut future = Box::pin(AsyncClient::connect(ClientConfig::default()));
+    assert!(matches!(
+        poll_once(Pin::as_mut(&mut future)),
+        Poll::Ready(Err(ClientError::NoTokioRuntime))
+    ));
+}
+
+#[test]
+fn recv_before_open_fails_on_first_poll_without_socket_readiness() {
+    runtime().block_on(async {
+        let remote = SocketAddr::from(([127, 0, 0, 1], 2112));
+        let mut client = AsyncClient::connect(config(remote, None, 0)).await.unwrap();
+
+        assert!(matches!(
+            poll_recv_once(&mut client),
+            Poll::Ready(Err(ClientError::NotOpen))
+        ));
+    });
+}
+
+#[test]
+fn ignored_open_traffic_cannot_extend_the_absolute_attempt_deadline() {
+    let server = start_server(move |socket, tx| {
+        let (first_packet, peer) = recv_packet(&socket, &tx);
+        let (params, _) = open_request(&first_packet, None);
+        socket
+            .set_read_timeout(Some(Duration::from_millis(10)))
+            .unwrap();
+        let started = Instant::now();
+        let mut next_noise = started;
+        let mut buffer = [0_u8; 512];
+
+        while started.elapsed() < Duration::from_millis(350) {
+            let now = Instant::now();
+            if now >= next_noise {
+                socket.send_to(&[0_u8], peer).unwrap();
+                next_noise = now + Duration::from_millis(25);
+            }
+            match socket.recv_from(&mut buffer) {
+                Ok((len, second_peer)) => {
+                    tx.send(buffer[..len].to_vec()).unwrap();
+                    send_open_reply(
+                        &socket,
+                        second_peer,
+                        params,
+                        None,
+                        flags::FLAG_OPEN | flags::FLAG_REPLY,
+                        TOKEN,
+                    );
+                    return;
+                }
+                Err(error)
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut
+                    ) => {}
+                Err(error) => panic!("{error}"),
+            }
+        }
+        panic!("second open request did not arrive after the first absolute deadline");
+    });
+    let mut client_config = config(server.addr, None, 0);
+    client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)];
+
+    runtime().block_on(async {
+        let mut client = AsyncClient::connect(client_config).await.unwrap();
+        assert!(matches!(
+            client.open().await.unwrap(),
+            OpenOutcome::Started { .. }
+        ));
+    });
+    assert_eq!(server.finish().len(), 2);
+}
+
+#[test]
+fn authenticated_rejection_is_terminal_without_retry() {
+    let server = start_server(move |socket, tx| {
+        let (open_packet, peer) = recv_packet(&socket, &tx);
+        let (params, _) = open_request(&open_packet, None);
+        send_open_reply(
+            &socket,
+            peer,
+            params,
+            None,
+            flags::FLAG_OPEN | flags::FLAG_REPLY | flags::FLAG_CLOSE,
+            0,
+        );
+    });
+    let mut client_config = config(server.addr, None, 0);
+    client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)];
+
+    runtime().block_on(async {
+        let mut client = AsyncClient::connect(client_config).await.unwrap();
+        assert!(matches!(
+            client.open().await,
+            Err(ClientError::ServerRejected)
+        ));
+    });
+    assert_eq!(server.finish().len(), 1);
+}
+
+#[test]
+fn recv_after_local_close_and_no_test_fails_on_first_poll() {
+    let close_server = start_server(move |socket, tx| {
+        open_session(&socket, &tx, None);
+        let (close_packet, _) = recv_packet(&socket, &tx);
+        assert_eq!(close_request_token(&close_packet, None), TOKEN);
+    });
+    runtime().block_on(async {
+        let mut client = opened_client(close_server.addr, 0).await;
+        client.close().await.unwrap();
+        assert!(matches!(
+            poll_recv_once(&mut client),
+            Poll::Ready(Err(ClientError::AlreadyClosed))
+        ));
+    });
+    assert_eq!(close_server.finish().len(), 2);
+
+    let no_test_server = start_no_test_server();
+    runtime().block_on(async {
+        let mut client_config = config(no_test_server.addr, None, 0);
+        client_config.run_mode = RunMode::NoTest;
+        let mut client = AsyncClient::connect(client_config).await.unwrap();
+        client.open().await.unwrap();
+        assert!(matches!(
+            poll_recv_once(&mut client),
+            Poll::Ready(Err(ClientError::AlreadyCompleted))
+        ));
+    });
+    assert_eq!(no_test_server.finish().len(), 1);
+}
+
+#[test]
+fn authenticated_async_lifecycle_negotiates_dscp() {
+    let key = b"async-lifecycle-key".to_vec();
+    let server = InTreeServer::start(ServerConfig::default().with_hmac_key(key.clone()));
+    runtime().block_on(async {
+        let mut client = AsyncClient::connect(config(server.addr, Some(key), 46))
+            .await
+            .unwrap();
+        let opened = client.open().await.unwrap();
+        assert_eq!(open_negotiated(&opened).params.dscp, 184);
+        client.send_probe().await.unwrap();
+        assert!(matches!(client.recv().await.unwrap().as_slice(), [ClientEvent::EchoReply { seq: 0, .. }]));
+        assert!(matches!(client.close().await.unwrap().as_slice(), [ClientEvent::SessionClosed { token, .. }] if *token == open_token(&opened)));
+    });
+}
+
+#[test]
+fn blocking_and_async_peer_close_are_semantically_equivalent() {
+    let blocking_peer_close_server = start_peer_close_server();
+    let mut blocking_peer_close =
+        Client::connect(config(blocking_peer_close_server.addr, None, 0)).unwrap();
+    blocking_peer_close.open().unwrap();
+    blocking_peer_close.send_probe().unwrap();
+    let blocking_peer_close_events = blocking_peer_close.recv_once().unwrap();
+    assert!(blocking_peer_close.is_peer_closed());
+    assert_eq!(blocking_peer_close_server.finish().len(), 2);
+
+    let async_peer_close_server = start_peer_close_server();
+    let async_peer_close_events = runtime().block_on(async {
+        let mut client = AsyncClient::connect(config(async_peer_close_server.addr, None, 0))
+            .await
+            .unwrap();
+        client.open().await.unwrap();
+        client.send_probe().await.unwrap();
+        let events = client.recv().await.unwrap();
+        assert!(client.is_peer_closed());
+        events
+    });
+    assert_eq!(async_peer_close_server.finish().len(), 2);
+    assert_eq!(
+        blocking_peer_close_events.len(),
+        async_peer_close_events.len()
+    );
+    for (blocking, asynchronous) in blocking_peer_close_events
+        .iter()
+        .zip(&async_peer_close_events)
+    {
+        assert_matching_event_shape(blocking, asynchronous);
+    }
+}
+
+fn open_negotiated(outcome: &OpenOutcome) -> &irtt_client::NegotiatedParams {
+    match outcome {
+        OpenOutcome::Started { negotiated, .. }
+        | OpenOutcome::NoTestCompleted { negotiated, .. } => negotiated,
+    }
+}
+
+fn open_token(outcome: &OpenOutcome) -> u64 {
+    match outcome {
+        OpenOutcome::Started { token, .. } => *token,
+        OpenOutcome::NoTestCompleted { .. } => panic!("no-test open has no session token"),
+    }
+}
+
+fn assert_matching_event_shape(blocking: &ClientEvent, asynchronous: &ClientEvent) {
+    assert_eq!(event_shape(blocking), event_shape(asynchronous));
+}
+
+fn event_shape(event: &ClientEvent) -> (&'static str, u64, usize) {
+    match event {
+        ClientEvent::EchoSent { seq, bytes, .. } => ("sent", u64::from(*seq), *bytes),
+        ClientEvent::EchoReply { seq, bytes, .. } => ("reply", u64::from(*seq), *bytes),
+        ClientEvent::SessionClosed { .. } => ("closed", 0, 0),
+        other => panic!("unexpected equivalence event: {other:?}"),
+    }
+}
+
+/// A peer that ignores the first `ignored_requests` datagrams before serving a
+/// compliant open, one echo, and one close.
+///
+/// Silence is the point here, so this stays a narrow fake peer rather than a
+/// real `irtt-server`: a compliant server has no way to be deliberately mute
+/// for exactly one open attempt.
+fn start_silent_then_open_server(ignored_requests: usize) -> TestServer {
+    start_server(move |socket, tx| {
+        for _ in 0..ignored_requests {
+            let _ = recv_packet(&socket, &tx);
+        }
+        let (params, peer) = open_session(&socket, &tx, None);
+        let (probe_packet, _) = recv_packet(&socket, &tx);
+        let sequence = echo_request_sequence(&probe_packet, None);
+        socket
+            .send_to(
+                &echo_reply(&params, sequence, TOKEN, flags::FLAG_REPLY, None),
+                peer,
+            )
+            .unwrap();
+        let (close_packet, _) = recv_packet(&socket, &tx);
+        assert_eq!(close_request_token(&close_packet, None), TOKEN);
+    })
+}
+
+/// Stable name for a `ClientError` variant, for comparing what each driver
+/// promises rather than how it produced it.
+fn error_name(error: &ClientError) -> &'static str {
+    match error {
+        ClientError::OpenTimeout => "open timeout",
+        ClientError::NotOpen => "not open",
+        ClientError::AlreadyOpen => "already open",
+        ClientError::AlreadyClosed => "already closed",
+        ClientError::AlreadyCompleted => "already completed",
+        ClientError::ServerRejected => "server rejected",
+        other => panic!("unexpected conformance error: {other:?}"),
+    }
+}
+
+fn probe_sequences(events: &[ClientEvent]) -> Vec<u32> {
+    events
+        .iter()
+        .filter_map(|event| match event {
+            ClientEvent::EchoSent { seq, .. } => Some(*seq),
+            _ => None,
+        })
+        .collect()
+}
+
+fn reply_sequences(events: &[ClientEvent]) -> Vec<u32> {
+    events
+        .iter()
+        .filter_map(|event| match event {
+            ClientEvent::EchoReply { seq, .. } => Some(*seq),
+            _ => None,
+        })
+        .collect()
+}
+
+#[test]
+fn blocking_and_async_roll_back_a_failed_open_and_accept_a_retry() {
+    // One open attempt is ignored, so the first open times out and the second
+    // must be able to open the same connected client.
+    let blocking_server = start_silent_then_open_server(1);
+    let mut blocking = Client::connect(config(blocking_server.addr, None, 0)).unwrap();
+    assert!(!blocking.has_pending_probes());
+    let blocking_first = blocking.open().unwrap_err();
+    assert!(!blocking.has_pending_probes());
+    let blocking_second = blocking.open().unwrap();
+    assert!(!blocking.has_pending_probes());
+    let blocking_sent = blocking.send_probe().unwrap();
+    let blocking_reply = blocking.recv_once().unwrap();
+    let blocking_close = blocking.close().unwrap();
+    assert_eq!(blocking_server.finish().len(), 4);
+
+    let async_server = start_silent_then_open_server(1);
+    let (async_first, async_second, async_sent, async_reply, async_close) =
+        runtime().block_on(async {
+            let mut client = AsyncClient::connect(config(async_server.addr, None, 0))
+                .await
+                .unwrap();
+            assert!(!client.has_pending_probes());
+            let first = client.open().await.unwrap_err();
+            assert!(!client.has_pending_probes());
+            let second = client.open().await.unwrap();
+            assert!(!client.has_pending_probes());
+            let sent = client.send_probe().await.unwrap();
+            let reply = client.recv().await.unwrap();
+            let closed = client.close().await.unwrap();
+            (first, second, sent, reply, closed)
+        });
+    assert_eq!(async_server.finish().len(), 4);
+
+    // Both drivers promise the same failure, and neither leaves the client in a
+    // state that refuses the retry.
+    assert_eq!(error_name(&blocking_first), "open timeout");
+    assert_eq!(error_name(&blocking_first), error_name(&async_first));
+    assert!(matches!(blocking_second, OpenOutcome::Started { .. }));
+    assert!(matches!(async_second, OpenOutcome::Started { .. }));
+    assert_eq!(
+        open_negotiated(&blocking_second),
+        open_negotiated(&async_second)
+    );
+    assert_eq!(open_token(&blocking_second), open_token(&async_second));
+
+    // The retried session starts a fresh probe sequence in both drivers.
+    assert_eq!(probe_sequences(&blocking_sent), [0]);
+    assert_eq!(probe_sequences(&async_sent), [0]);
+    assert_eq!(reply_sequences(&blocking_reply), [0]);
+    assert_eq!(reply_sequences(&async_reply), [0]);
+    assert_eq!(blocking_sent.len(), async_sent.len());
+    assert_eq!(blocking_reply.len(), async_reply.len());
+    assert_eq!(blocking_close.len(), async_close.len());
+    assert_matching_event_shape(&blocking_close[0], &async_close[0]);
+}
+
+#[test]
+fn blocking_and_async_complete_every_caller_paced_probe() {
+    const PROBES: u32 = 3;
+    fn caller_config(addr: SocketAddr) -> ClientConfig {
+        ClientConfig {
+            duration: Some(Duration::from_millis(1)),
+            interval: Duration::from_secs(1),
+            ..config(addr, None, 0)
+        }
+    }
+
+    let blocking_server = InTreeServer::start(ServerConfig::default());
+    let mut blocking = Client::connect(caller_config(blocking_server.addr)).unwrap();
+    let outcome = blocking.open().unwrap();
+    assert_eq!(
+        blocking.negotiated_params(),
+        Some(open_negotiated(&outcome))
+    );
+    assert!(!blocking.has_pending_probes());
+    // Oversleeping cannot invalidate this: low-level sends remain caller-owned
+    // after the negotiated duration. The compliant server has no duration cap.
+    thread::sleep(Duration::from_millis(2));
+    let mut blocking_sent = Vec::new();
+    let mut blocking_replies = Vec::new();
+    for _ in 0..PROBES {
+        blocking_sent.extend(blocking.send_probe().unwrap());
+        assert!(blocking.has_pending_probes());
+        assert!(blocking.next_probe_timeout_deadline().is_some());
+        blocking_replies.extend(blocking.recv_once().unwrap());
+        assert!(!blocking.has_pending_probes());
+    }
+    blocking.close().unwrap();
+    drop(blocking_server);
+
+    let async_server = InTreeServer::start(ServerConfig::default());
+    let (async_sent, async_replies) = runtime().block_on(async {
+        let mut client = AsyncClient::connect(caller_config(async_server.addr))
+            .await
+            .unwrap();
+        let outcome = client.open().await.unwrap();
+        assert_eq!(client.negotiated_params(), Some(open_negotiated(&outcome)));
+        tokio::time::sleep(Duration::from_millis(2)).await;
+        let mut sent = Vec::new();
+        let mut replies = Vec::new();
+        for _ in 0..PROBES {
+            sent.extend(client.send_probe().await.unwrap());
+            assert!(client.has_pending_probes());
+            assert!(client.next_probe_timeout_deadline().is_some());
+            replies.extend(client.recv().await.unwrap());
+            assert!(!client.has_pending_probes());
+        }
+        client.close().await.unwrap();
+        (sent, replies)
+    });
+    drop(async_server);
+
+    let expected: Vec<u32> = (0..PROBES).collect();
+    assert_eq!(probe_sequences(&blocking_sent), expected);
+    assert_eq!(probe_sequences(&async_sent), expected);
+    assert_eq!(reply_sequences(&blocking_replies), expected);
+    assert_eq!(reply_sequences(&async_replies), expected);
+    // Compare the whole emitted streams, not just the probe and reply events
+    // the sequence helpers keep: an extra event on either side is a difference.
+    assert_eq!(blocking_sent.len(), async_sent.len());
+    assert_eq!(blocking_replies.len(), async_replies.len());
+    for event in blocking_sent.iter().chain(&async_sent) {
+        assert!(matches!(
+            event,
+            ClientEvent::EchoSent {
+                scheduled_at: None,
+                timer_error: None,
+                ..
+            }
+        ));
+    }
+    for (blocking, asynchronous) in blocking_sent.iter().zip(&async_sent) {
+        assert_matching_event_shape(blocking, asynchronous);
+    }
+    for (blocking, asynchronous) in blocking_replies.iter().zip(&async_replies) {
+        assert_matching_event_shape(blocking, asynchronous);
+    }
+}
+
+#[test]
+fn ignored_open_traffic_uses_one_request_and_deadline_per_attempt() {
+    let server = start_server(move |socket, tx| {
+        for _ in 0..2 {
+            let (_, peer) = recv_packet(&socket, &tx);
+            for _ in 0..3 {
+                socket.send_to(&[0_u8], peer).unwrap();
+            }
+        }
+    });
+    let mut client_config = config(server.addr, None, 0);
+    client_config.open_timeouts = vec![Duration::from_millis(200), Duration::from_millis(200)];
+
+    runtime().block_on(async {
+        let mut client = AsyncClient::connect(client_config).await.unwrap();
+        assert!(matches!(client.open().await, Err(ClientError::OpenTimeout)));
+    });
+    assert_eq!(server.finish().len(), 2);
+}

crates/irtt-integration-tests/tests/blocking_managed.rs

@@ -0,0 +1,340 @@
+use std::{
+    net::{SocketAddr, UdpSocket},
+    sync::{Arc, Condvar, Mutex},
+    thread::{self, JoinHandle},
+    time::{Duration, Instant},
+};
+
+use irtt_proto::{
+    decode_request, encode_echo_reply, encode_open_reply, flags, DecodedRequestKind, EchoReply,
+    OpenReply, Params, ReceivedStats, StampAt, TimestampFields,
+};
+use tokio::{runtime::Builder, time::timeout};
+
+use irtt_client::managed::{
+    BlockingManagedClient, ManagedClientConfig, ManagedCompletionPolicy, ManagedEndReason,
+    ManagedEvent, ManagedLifecycle, ManagedPacing, ManagedTargetConfig, ManagedTargetEndReason,
+};
+use irtt_client::ClientConfig;
+
+const TOKEN: u64 = 0x1234_5678_90ab_cdef;
+
+struct Signal {
+    value: Mutex<bool>,
+    ready: Condvar,
+}
+
+impl Signal {
+    fn set(&self) {
+        *self.value.lock().unwrap() = true;
+        self.ready.notify_all();
+    }
+
+    fn wait(&self) {
+        let value = self.value.lock().unwrap();
+        let (value, timeout) = self
+            .ready
+            .wait_timeout_while(value, Duration::from_secs(2), |value| !*value)
+            .unwrap();
+        assert!(*value && !timeout.timed_out());
+    }
+}
+
+struct EchoServer {
+    addr: SocketAddr,
+    probe: Arc<Signal>,
+    close: Arc<Signal>,
+    worker: JoinHandle<()>,
+}
+
+impl EchoServer {
+    fn start() -> Self {
+        let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+        socket
+            .set_read_timeout(Some(Duration::from_secs(3)))
+            .unwrap();
+        let addr = socket.local_addr().unwrap();
+        let probe = Arc::new(Signal {
+            value: Mutex::new(false),
+            ready: Condvar::new(),
+        });
+        let close = Arc::new(Signal {
+            value: Mutex::new(false),
+            ready: Condvar::new(),
+        });
+        let worker_probe = Arc::clone(&probe);
+        let worker_close = Arc::clone(&close);
+        let worker = thread::spawn(move || {
+            let mut buffer = [0_u8; 8192];
+            let (open_len, peer) = socket.recv_from(&mut buffer).unwrap();
+            let negotiated = match decode_request(&buffer[..open_len]).unwrap().kind {
+                DecodedRequestKind::Open { params, .. } => Params::decode(params).unwrap(),
+                other => panic!("expected an open request, got {other:?}"),
+            };
+            let reply = encode_open_reply(
+                &OpenReply {
+                    flags: flags::FLAG_OPEN | flags::FLAG_REPLY,
+                    token: TOKEN,
+                    params: negotiated.clone(),
+                },
+                None,
+            )
+            .unwrap();
+            socket.send_to(&reply, peer).unwrap();
+
+            loop {
+                let (len, packet_peer) = socket.recv_from(&mut buffer).unwrap();
+                let Ok(request) = decode_request(&buffer[..len]) else {
+                    continue;
+                };
+                match request.kind {
+                    DecodedRequestKind::Echo { sequence, .. } => {
+                        worker_probe.set();
+                        let reply = encode_echo_reply(
+                            &EchoReply {
+                                flags: flags::FLAG_REPLY,
+                                token: TOKEN,
+                                sequence,
+                                recv_count: None,
+                                recv_window: None,
+                                timestamps: TimestampFields::default(),
+                                payload: Vec::new(),
+                            },
+                            &negotiated,
+                            None,
+                        )
+                        .unwrap();
+                        socket.send_to(&reply, packet_peer).unwrap();
+                    }
+                    DecodedRequestKind::Close { .. } => {
+                        worker_close.set();
+                        return;
+                    }
+                    DecodedRequestKind::Open { .. } => {}
+                }
+            }
+        });
+        Self {
+            addr,
+            probe,
+            close,
+            worker,
+        }
+    }
+
+    fn wait_probe(&self) {
+        self.probe.wait();
+    }
+
+    fn wait_close(&self) {
+        self.close.wait();
+    }
+
+    fn finish(self) {
+        self.worker.join().unwrap();
+    }
+}
+
+fn test_runtime() -> tokio::runtime::Runtime {
+    Builder::new_current_thread()
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap()
+}
+
+fn config(completion: ManagedCompletionPolicy) -> ManagedClientConfig {
+    ManagedClientConfig {
+        client: ClientConfig {
+            duration: Some(Duration::from_millis(120)),
+            interval: Duration::from_millis(20),
+            probe_timeout: Duration::from_millis(40),
+            open_timeouts: vec![Duration::from_millis(200)],
+            received_stats: ReceivedStats::None,
+            stamp_at: StampAt::None,
+            ..ClientConfig::default()
+        },
+        pacing: ManagedPacing::Burst,
+        completion,
+        final_drain: Duration::from_millis(5),
+        ..ManagedClientConfig::default()
+    }
+}
+
+fn target(server: &EchoServer) -> ManagedTargetConfig {
+    ManagedTargetConfig::new("target", server.addr.to_string())
+}
+
+fn assert_stopped(outcome: &irtt_client::managed::ManagedOutcome) {
+    assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+}
+
+#[test]
+fn starts_without_caller_runtime_and_finishes_naturally() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+
+    server.wait_probe();
+    let outcome = owner.join().unwrap();
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+    assert_eq!(outcome.recent_target_outcomes.len(), 1);
+    assert_eq!(
+        outcome.recent_target_outcomes[0].end_reason,
+        ManagedTargetEndReason::TestComplete
+    );
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn subscribed_start_captures_events_from_a_short_run() {
+    let server = EchoServer::start();
+    let (owner, mut events) = BlockingManagedClient::start_with_subscription(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+
+    server.wait_probe();
+    let outcome = owner.join().unwrap();
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+
+    let mut saw_started = false;
+    let mut saw_finished = false;
+    let mut saw_completed = false;
+    while let Ok(event) = events.try_recv() {
+        match event {
+            ManagedEvent::Started => saw_started = true,
+            ManagedEvent::TargetFinished { .. } => saw_finished = true,
+            ManagedEvent::Completed { .. } => saw_completed = true,
+            _ => {}
+        }
+    }
+    assert!(saw_started && saw_finished && saw_completed);
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn external_handle_stop_ends_with_graceful_close() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(
+        config(ManagedCompletionPolicy::ExplicitStop),
+        vec![target(&server)],
+    )
+    .unwrap();
+    let handle = owner.handle();
+
+    server.wait_probe();
+    test_runtime().block_on(async {
+        timeout(Duration::from_secs(2), handle.stop())
+            .await
+            .unwrap();
+    });
+    let outcome = owner.join().unwrap();
+    assert_stopped(&outcome);
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn dynamic_update_reaches_background_worker() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(config(ManagedCompletionPolicy::ExplicitStop), vec![])
+        .unwrap();
+    let handle = owner.handle();
+
+    let acknowledgement = handle
+        .update_targets(vec![target(&server)])
+        .unwrap()
+        .blocking_wait()
+        .unwrap();
+    assert_eq!(acknowledgement.sequence, 1);
+    server.wait_probe();
+    test_runtime().block_on(async {
+        timeout(Duration::from_secs(2), owner.handle().stop())
+            .await
+            .unwrap();
+    });
+    assert_stopped(&owner.join().unwrap());
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn dropping_external_handles_does_not_stop_owner() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+    drop(owner.handle());
+    drop(owner.handle());
+
+    server.wait_probe();
+    assert_eq!(
+        owner.join().unwrap().end_reason,
+        ManagedEndReason::TargetsComplete
+    );
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn dropping_owner_stops_and_joins_worker() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(
+        config(ManagedCompletionPolicy::ExplicitStop),
+        vec![target(&server)],
+    )
+    .unwrap();
+    let handle = owner.handle();
+
+    server.wait_probe();
+    drop(owner);
+    let status = handle.status();
+    assert_eq!(status.lifecycle, ManagedLifecycle::Completed);
+    assert_eq!(
+        status.final_outcome.as_ref().unwrap().end_reason,
+        ManagedEndReason::StopRequested
+    );
+    server.wait_close();
+    server.finish();
+}
+
+#[test]
+fn drop_after_natural_completion_preserves_outcome() {
+    let server = EchoServer::start();
+    let owner = BlockingManagedClient::start(
+        config(ManagedCompletionPolicy::FinishWhenQuiescent),
+        vec![target(&server)],
+    )
+    .unwrap();
+    let handle = owner.handle();
+    server.wait_probe();
+    server.wait_close();
+    let deadline = Instant::now() + Duration::from_secs(2);
+    let status = loop {
+        let status = handle.status();
+        if status.lifecycle == ManagedLifecycle::Completed && status.final_outcome.is_some() {
+            break status;
+        }
+        assert!(Instant::now() < deadline, "managed worker did not complete");
+        thread::sleep(Duration::from_millis(5));
+    };
+    assert_eq!(
+        status.final_outcome.as_ref().unwrap().end_reason,
+        ManagedEndReason::TargetsComplete
+    );
+    drop(owner);
+    assert_eq!(
+        handle.status().final_outcome.as_ref().unwrap().end_reason,
+        ManagedEndReason::TargetsComplete
+    );
+    server.finish();
+}

crates/irtt-integration-tests/tests/managed_behavior.rs

@@ -0,0 +1,248 @@
+mod support;
+
+use std::{collections::HashMap, time::Duration};
+
+use irtt_client::{managed::*, ClientConfig, ClientEvent};
+use irtt_server::ServerConfig;
+use support::InTreeServer;
+use tokio::time::timeout;
+
+// A real service and public events protect cadence/fairness without depending
+// on a scheduler, a send cursor, or a private event-publication hook.
+#[tokio::test(flavor = "current_thread")]
+async fn both_pacing_modes_serve_every_target_on_its_negotiated_grid() {
+    timeout(Duration::from_secs(10), async {
+        let server = InTreeServer::start(ServerConfig::default());
+        for pacing in [ManagedPacing::Burst, ManagedPacing::Staggered] {
+            let mut observed_pacing = false;
+            for _ in 0..3 {
+                let config = ManagedClientConfig {
+                    client: ClientConfig {
+                        duration: Some(Duration::from_millis(600)),
+                        interval: Duration::from_millis(100),
+                        probe_timeout: Duration::from_millis(100),
+                        open_timeouts: vec![Duration::from_millis(200)],
+                        ..ClientConfig::default()
+                    },
+                    pacing,
+                    ..ManagedClientConfig::default()
+                };
+                let targets = ["first", "second"]
+                    .map(|id| ManagedTargetConfig::new(id, server.addr.to_string()));
+                let (task, handle) = ManagedClient::task(config, targets.into()).unwrap();
+                let mut events = handle.subscribe().unwrap();
+                let run = tokio::spawn(task);
+                let mut sends: HashMap<TargetInstance, Vec<_>> = HashMap::new();
+                let mut actual_sends = Vec::new();
+                while let Ok(event) = events.recv().await {
+                    if let ManagedEvent::Client {
+                        target,
+                        event:
+                            ClientEvent::EchoSent {
+                                scheduled_at,
+                                sent_at,
+                                ..
+                            },
+                    } = event
+                    {
+                        let scheduled = scheduled_at.expect("managed probes expose their cadence");
+                        assert!(sent_at.mono >= scheduled);
+                        actual_sends.push((target.clone(), sent_at.mono));
+                        let previous = sends.entry(target).or_default();
+                        if let Some(last) = previous.last() {
+                            let elapsed: Duration = scheduled.duration_since(*last);
+                            assert!(!elapsed.is_zero());
+                            assert_eq!(
+                                elapsed.as_nanos() % Duration::from_millis(100).as_nanos(),
+                                0
+                            );
+                        }
+                        previous.push(scheduled);
+                    }
+                }
+                let outcome = run.await.unwrap();
+                assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+                assert_eq!(outcome.successful_target_outcomes, 2);
+                assert_eq!(sends.len(), 2);
+                for target in outcome.recent_target_outcomes.iter() {
+                    assert!(sends[&target.target].len() >= 2);
+                    assert!(target.replies_received >= 2);
+                }
+                // Use only the initial pair: later stagger gates can legitimately
+                // compress after a delayed wake. Startup may instead serve one
+                // target twice before the other opens, so retry a fresh run then.
+                if actual_sends.len() < 2 || actual_sends[0].0 == actual_sends[1].0 {
+                    continue;
+                }
+                let gap = actual_sends[1].1.duration_since(actual_sends[0].1);
+                match pacing {
+                    ManagedPacing::Staggered => {
+                        assert!(
+                            gap >= Duration::from_millis(50),
+                            "first stagger gate was bypassed: {gap:?}"
+                        );
+                        observed_pacing = true;
+                    }
+                    ManagedPacing::Burst => {
+                        // A pause between burst sends makes this observation
+                        // inconclusive; it must not fail an otherwise valid run.
+                        observed_pacing = gap < Duration::from_millis(25);
+                    }
+                }
+                if observed_pacing {
+                    break;
+                }
+            }
+            assert!(
+                observed_pacing,
+                "could not observe {pacing:?} startup pacing in three fresh runs"
+            );
+        }
+    })
+    .await
+    .expect("managed pacing did not complete");
+}
+
+// Updates, receipts, generations and cleanup are observable through the control
+// handle and event subscription; no task internals or artificial races are needed.
+#[tokio::test(flavor = "current_thread")]
+async fn target_updates_preserve_identical_sessions_and_retire_removed_generations() {
+    timeout(Duration::from_secs(5), async {
+        let server = InTreeServer::start(ServerConfig::default());
+        let config = ManagedClientConfig {
+            client: ClientConfig {
+                duration: None,
+                interval: Duration::from_millis(20),
+                probe_timeout: Duration::from_millis(100),
+                open_timeouts: vec![Duration::from_millis(200)],
+                ..ClientConfig::default()
+            },
+            completion: ManagedCompletionPolicy::ExplicitStop,
+            outcome_history_limit: 1,
+            ..ManagedClientConfig::default()
+        };
+        let configured = ManagedTargetConfig::new("one", server.addr.to_string());
+        let (task, handle) = ManagedClient::task(config, vec![]).unwrap();
+        let mut events = handle.subscribe().unwrap();
+        let run = tokio::spawn(task);
+        let added = handle
+            .update_targets(vec![configured.clone()])
+            .unwrap()
+            .await
+            .unwrap();
+        let first = added.status.targets[0].target.clone();
+        loop {
+            if matches!(events.recv().await.unwrap(), ManagedEvent::Client {
+                target, event: ClientEvent::EchoReply { .. }
+            } if target == first)
+            {
+                break;
+            }
+        }
+        let identical = handle
+            .update_targets(vec![configured.clone()])
+            .unwrap()
+            .await
+            .unwrap();
+        assert_eq!(identical.status.targets.len(), 1);
+        assert_eq!(identical.status.targets[0].target, first);
+        loop {
+            match events.recv().await.unwrap() {
+                ManagedEvent::Client {
+                    target,
+                    event: ClientEvent::SessionStarted { .. },
+                } if target == first => panic!("identical update reopened the session"),
+                ManagedEvent::Client {
+                    target,
+                    event: ClientEvent::EchoReply { .. },
+                } if target == first => break,
+                _ => {}
+            }
+        }
+        handle.update_targets(vec![]).unwrap().await.unwrap();
+        loop {
+            if let ManagedEvent::TargetFinished { outcome } = events.recv().await.unwrap() {
+                let status = handle.status();
+                assert_eq!(outcome.target, first);
+                assert_eq!(outcome.end_reason, ManagedTargetEndReason::Removed);
+                assert_eq!(outcome.cleanup_failure, None);
+                assert_eq!(status.total_target_outcomes, 1);
+                assert_eq!(status.successful_target_outcomes, 1);
+                assert_eq!(
+                    status.recent_target_outcomes.as_ref(),
+                    &[(*outcome).clone()]
+                );
+                assert!(status.targets.iter().all(|target| target.target != first
+                    || target.lifecycle == ManagedTargetLifecycle::Terminal));
+                break;
+            }
+        }
+        let readded = handle
+            .update_targets(vec![configured])
+            .unwrap()
+            .await
+            .unwrap();
+        let second = readded
+            .status
+            .targets
+            .iter()
+            .find(|target| target.desired)
+            .unwrap()
+            .target
+            .clone();
+        assert!(second.generation > first.generation);
+        loop {
+            if matches!(events.recv().await.unwrap(), ManagedEvent::Client {
+                target, event: ClientEvent::EchoReply { .. }
+            } if target == second)
+            {
+                break;
+            }
+        }
+        let stop = handle.stop();
+        loop {
+            match events.recv().await.unwrap() {
+                ManagedEvent::TargetFinished { outcome } => {
+                    let status = handle.status();
+                    assert_eq!(outcome.target, second);
+                    assert_eq!(status.total_target_outcomes, 2);
+                    assert_eq!(
+                        status.recent_target_outcomes.as_ref(),
+                        &[(*outcome).clone()]
+                    );
+                    assert!(status.targets.iter().all(|target| target.target != second
+                        || target.lifecycle == ManagedTargetLifecycle::Terminal));
+                }
+                ManagedEvent::Completed { outcome } => {
+                    let status = handle.status();
+                    assert_eq!(status.lifecycle, ManagedLifecycle::Completed);
+                    assert_eq!(status.final_outcome.as_ref(), Some(&outcome));
+                    break;
+                }
+                _ => {}
+            }
+        }
+        let outcome = run.await.unwrap();
+        stop.await;
+        assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+        assert_eq!(outcome.total_target_outcomes, 2);
+        assert_eq!(outcome.successful_target_outcomes, 2);
+        assert_eq!(outcome.failed_target_outcomes, 0);
+        assert_eq!(outcome.discarded_target_outcomes, 1);
+        assert_eq!(outcome.recent_target_outcomes.len(), 1);
+        let status = handle.status();
+        assert_eq!(status.total_target_outcomes, 2);
+        assert_eq!(status.successful_target_outcomes, 2);
+        assert_eq!(status.failed_target_outcomes, 0);
+        assert_eq!(status.discarded_target_outcomes, 1);
+        assert_eq!(status.recent_target_outcomes.len(), 1);
+        assert_eq!(status.recent_target_outcomes[0].target, second);
+        assert!(outcome.recent_target_outcomes.iter().any(|target| {
+            target.target == second
+                && target.end_reason == ManagedTargetEndReason::Stopped
+                && target.cleanup_failure.is_none()
+        }));
+    })
+    .await
+    .expect("managed target updates did not complete");
+}

crates/irtt-integration-tests/tests/managed_lifecycle.rs

@@ -0,0 +1,506 @@
+use std::{
+    future::Future,
+    net::{SocketAddr, UdpSocket},
+    pin::Pin,
+    sync::{Arc, Mutex},
+    task::{Context, Poll, Waker},
+    thread::{self, JoinHandle},
+    time::{Duration, Instant},
+};
+
+use irtt_proto::{
+    decode_request, encode_echo_reply, encode_open_reply, flags, verify_packet_hmac,
+    DecodedRequestKind, EchoReply, OpenReply, Params, ReceivedStats, StampAt, TimestampFields,
+};
+use tokio::runtime::{Builder, Runtime};
+
+use irtt_client::{managed::*, ClientConfig};
+use irtt_client::{ClientAuthConfig, RunMode};
+use tokio::sync::broadcast;
+
+const TOKEN: u64 = 0x1234_5678_90ab_cdef;
+
+#[derive(Clone, Copy)]
+enum ServerBehavior {
+    Echo,
+    NoTest,
+    PeerClose,
+    DelayedEcho(Duration),
+}
+
+#[derive(Clone, Debug)]
+enum PacketKind {
+    Open,
+    Probe,
+    Close,
+}
+
+#[derive(Clone, Debug)]
+struct PacketRecord {
+    kind: PacketKind,
+    at: Instant,
+}
+
+struct TestServer {
+    addr: SocketAddr,
+    records: Arc<Mutex<Vec<PacketRecord>>>,
+    thread: JoinHandle<()>,
+}
+
+impl TestServer {
+    fn finish(self) -> Vec<PacketRecord> {
+        self.thread.join().unwrap();
+        self.records.lock().unwrap().clone()
+    }
+}
+
+/// Admits a request the way a compliant server does: structural decode first,
+/// then HMAC presence policy, then authentication. Returns `None` for anything
+/// a server would silently discard.
+fn decode_server_request<'a>(
+    packet: &'a [u8],
+    key: Option<&[u8]>,
+) -> Option<DecodedRequestKind<'a>> {
+    let request = decode_request(packet).ok()?;
+    if request.hmac_present != key.is_some() {
+        return None;
+    }
+    if let Some(key) = key {
+        verify_packet_hmac(key, packet).ok()?;
+    }
+    Some(request.kind)
+}
+
+fn decode_open_params(packet: &[u8], key: Option<&[u8]>) -> Params {
+    match decode_server_request(packet, key) {
+        Some(DecodedRequestKind::Open { params, .. }) => Params::decode(params).unwrap(),
+        other => panic!("expected an authenticated open request, got {other:?}"),
+    }
+}
+
+fn start_server(behavior: ServerBehavior, key: Option<Vec<u8>>) -> TestServer {
+    let socket = UdpSocket::bind("127.0.0.1:0").unwrap();
+    socket
+        .set_read_timeout(Some(Duration::from_secs(3)))
+        .unwrap();
+    let addr = socket.local_addr().unwrap();
+    let records = Arc::new(Mutex::new(Vec::new()));
+    let thread_records = Arc::clone(&records);
+    let thread = thread::spawn(move || {
+        let mut buffer = [0_u8; 8192];
+        let (open_len, peer) = socket.recv_from(&mut buffer).unwrap();
+        thread_records.lock().unwrap().push(PacketRecord {
+            kind: PacketKind::Open,
+            at: Instant::now(),
+        });
+        let negotiated = decode_open_params(&buffer[..open_len], key.as_deref());
+        let no_test = matches!(behavior, ServerBehavior::NoTest);
+        let open_reply = encode_open_reply(
+            &OpenReply {
+                flags: flags::FLAG_OPEN
+                    | flags::FLAG_REPLY
+                    | if no_test { flags::FLAG_CLOSE } else { 0 },
+                token: if no_test { 0 } else { TOKEN },
+                params: negotiated.clone(),
+            },
+            key.as_deref(),
+        )
+        .unwrap();
+        socket.send_to(&open_reply, peer).unwrap();
+        if no_test {
+            return;
+        }
+
+        loop {
+            let Ok((len, packet_peer)) = socket.recv_from(&mut buffer) else {
+                return;
+            };
+            let packet = &buffer[..len];
+            match decode_server_request(packet, key.as_deref()) {
+                Some(DecodedRequestKind::Echo { sequence, .. }) => {
+                    thread_records.lock().unwrap().push(PacketRecord {
+                        kind: PacketKind::Probe,
+                        at: Instant::now(),
+                    });
+                    if let ServerBehavior::DelayedEcho(delay) = behavior {
+                        thread::sleep(delay);
+                    }
+                    let peer_close = matches!(behavior, ServerBehavior::PeerClose);
+                    let reply = encode_echo_reply(
+                        &EchoReply {
+                            flags: flags::FLAG_REPLY
+                                | if peer_close { flags::FLAG_CLOSE } else { 0 },
+                            token: TOKEN,
+                            sequence,
+                            recv_count: None,
+                            recv_window: None,
+                            timestamps: TimestampFields::default(),
+                            payload: Vec::new(),
+                        },
+                        &negotiated,
+                        key.as_deref(),
+                    )
+                    .unwrap();
+                    socket.send_to(&reply, packet_peer).unwrap();
+                    if peer_close {
+                        return;
+                    }
+                }
+                Some(DecodedRequestKind::Close { .. }) => {
+                    thread_records.lock().unwrap().push(PacketRecord {
+                        kind: PacketKind::Close,
+                        at: Instant::now(),
+                    });
+                    return;
+                }
+                Some(DecodedRequestKind::Open { .. }) | None => {}
+            }
+        }
+    });
+    TestServer {
+        addr,
+        records,
+        thread,
+    }
+}
+
+fn runtime() -> Runtime {
+    Builder::new_current_thread()
+        .enable_io()
+        .enable_time()
+        .build()
+        .unwrap()
+}
+
+fn config(pacing: ManagedPacing) -> ManagedClientConfig {
+    ManagedClientConfig {
+        client: ClientConfig {
+            duration: Some(Duration::from_millis(130)),
+            interval: Duration::from_millis(60),
+            probe_timeout: Duration::from_millis(35),
+            open_timeouts: vec![Duration::from_millis(200)],
+            received_stats: ReceivedStats::None,
+            stamp_at: StampAt::None,
+            ..ClientConfig::default()
+        },
+        pacing,
+        final_drain: Duration::from_millis(5),
+        ..ManagedClientConfig::default()
+    }
+}
+
+fn target(id: &str, addr: SocketAddr) -> ManagedTargetConfig {
+    ManagedTargetConfig::new(id, addr.to_string())
+}
+
+fn poll_once<F: Future>(future: Pin<&mut F>) -> Poll<F::Output> {
+    Future::poll(future, &mut Context::from_waker(Waker::noop()))
+}
+
+fn probes(records: &[PacketRecord]) -> Vec<Instant> {
+    records
+        .iter()
+        .filter_map(|record| matches!(record.kind, PacketKind::Probe).then_some(record.at))
+        .collect()
+}
+
+fn has_close(records: &[PacketRecord]) -> bool {
+    records
+        .iter()
+        .any(|record| matches!(record.kind, PacketKind::Close))
+}
+
+#[test]
+fn construction_is_runtime_and_io_free() {
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![ManagedTargetConfig::new("dns", "no-such-host.invalid")],
+    )
+    .unwrap();
+    assert_eq!(handle.status().lifecycle, ManagedLifecycle::NotStarted);
+    drop(task);
+}
+
+#[test]
+fn quiescent_rejects_empty_initial_targets() {
+    assert!(matches!(
+        ManagedClient::task(config(ManagedPacing::Staggered), vec![]),
+        Err(ManagedConfigError::EmptyInitialTargets)
+    ));
+}
+
+#[test]
+fn explicit_empty_waits_for_stop() {
+    let mut config = config(ManagedPacing::Staggered);
+    config.completion = ManagedCompletionPolicy::ExplicitStop;
+    let (task, handle) = ManagedClient::task(config, vec![]).unwrap();
+    let outcome = runtime().block_on(async move {
+        let mut task = Box::pin(task);
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), task.as_mut())
+                .await
+                .is_err()
+        );
+        assert_eq!(handle.status().lifecycle, ManagedLifecycle::Running);
+        let receipt = handle.stop();
+        let outcome = task.await;
+        receipt.await;
+        outcome
+    });
+    assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+}
+
+#[test]
+fn extreme_event_capacity_is_rejected_without_panic() {
+    let mut managed = config(ManagedPacing::Staggered);
+    managed.completion = ManagedCompletionPolicy::ExplicitStop;
+    managed.event_capacity = usize::MAX;
+    assert!(matches!(
+        ManagedClient::task(managed, vec![]),
+        Err(ManagedConfigError::EventCapacityTooLarge {
+            configured: usize::MAX,
+            ..
+        })
+    ));
+}
+
+#[test]
+fn extreme_command_capacity_is_rejected_without_panic() {
+    let mut managed = config(ManagedPacing::Staggered);
+    managed.completion = ManagedCompletionPolicy::ExplicitStop;
+    managed.command_capacity = usize::MAX;
+    assert!(matches!(
+        ManagedClient::task(managed, vec![]),
+        Err(ManagedConfigError::CommandCapacityTooLarge {
+            configured: usize::MAX,
+            ..
+        })
+    ));
+}
+
+#[test]
+fn first_poll_without_runtime_fails_durably() {
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![target("one", "127.0.0.1:9".parse().unwrap())],
+    )
+    .unwrap();
+    let mut task = Box::pin(task);
+    let Poll::Ready(outcome) = poll_once(task.as_mut()) else {
+        panic!("task did not fail on its first poll")
+    };
+    assert_eq!(
+        outcome.end_reason,
+        ManagedEndReason::DriverFailed(ManagedDriverFailure::NoTokioRuntime)
+    );
+    assert_eq!(handle.status().lifecycle, ManagedLifecycle::Failed);
+}
+
+#[test]
+fn pre_poll_stop() {
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![ManagedTargetConfig::new("one", "no-such-host.invalid")],
+    )
+    .unwrap();
+    let mut receipt = Box::pin(handle.stop());
+    let mut task = Box::pin(task);
+    let Poll::Ready(outcome) = poll_once(task.as_mut()) else {
+        panic!("pre-stopped task did not complete")
+    };
+    assert!(poll_once(receipt.as_mut()).is_ready());
+    assert_eq!(outcome.end_reason, ManagedEndReason::StopRequested);
+}
+
+#[test]
+fn finite_immediate_replies_use_retained_drain_deadline() {
+    let server = start_server(ServerBehavior::Echo, None);
+    let mut managed = config(ManagedPacing::Staggered);
+    managed.client.probe_timeout = Duration::from_secs(3);
+    managed.final_drain = Duration::from_millis(30);
+    let (task, _) = ManagedClient::task(managed, vec![target("one", server.addr)]).unwrap();
+    let started_at = Instant::now();
+    let outcome = runtime().block_on(async {
+        tokio::time::timeout(Duration::from_secs(1), task)
+            .await
+            .expect("replied probes must not hold the drain until their obsolete timeout")
+    });
+    assert!(started_at.elapsed() < Duration::from_secs(1));
+    let records = server.finish();
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+    assert!(matches!(
+        outcome.recent_target_outcomes[0].end_reason,
+        ManagedTargetEndReason::TestComplete
+    ));
+    assert!(!probes(&records).is_empty() && has_close(&records));
+}
+
+#[test]
+fn no_test_target_completion() {
+    let server = start_server(ServerBehavior::NoTest, None);
+    let mut config = config(ManagedPacing::Staggered);
+    config.client.run_mode = RunMode::NoTest;
+    let (task, _) = ManagedClient::task(config, vec![target("no-test", server.addr)]).unwrap();
+    let outcome = runtime().block_on(task);
+    let records = server.finish();
+    assert!(matches!(
+        outcome.recent_target_outcomes[0].end_reason,
+        ManagedTargetEndReason::NoTestComplete
+    ));
+    assert!(probes(&records).is_empty() && !has_close(&records));
+}
+
+#[test]
+fn authenticated_peer_close_outcome() {
+    let key = b"managed-peer-close".to_vec();
+    let server = start_server(ServerBehavior::PeerClose, Some(key.clone()));
+    let mut configured = target("peer", server.addr);
+    configured.auth = Some(ClientAuthConfig {
+        hmac_key: Some(key),
+    });
+    let (task, _) =
+        ManagedClient::task(config(ManagedPacing::Staggered), vec![configured]).unwrap();
+    let outcome = runtime().block_on(task);
+    let records = server.finish();
+    assert_eq!(outcome.peer_closed_target_outcomes, 1);
+    assert!(matches!(
+        outcome.recent_target_outcomes[0].end_reason,
+        ManagedTargetEndReason::PeerClosed
+    ));
+    assert!(!has_close(&records));
+}
+
+#[test]
+fn static_multi_target_completion() {
+    let first = start_server(ServerBehavior::Echo, None);
+    let second = start_server(ServerBehavior::Echo, None);
+    let (task, _) = ManagedClient::task(
+        config(ManagedPacing::Burst),
+        vec![target("first", first.addr), target("second", second.addr)],
+    )
+    .unwrap();
+    let outcome = runtime().block_on(task);
+    assert_eq!(outcome.total_target_outcomes, 2);
+    assert_eq!(outcome.successful_target_outcomes, 2);
+    assert!(!probes(&first.finish()).is_empty());
+    assert!(!probes(&second.finish()).is_empty());
+}
+
+#[test]
+fn sibling_failure_isolation() {
+    let good = start_server(ServerBehavior::Echo, None);
+    let unused = UdpSocket::bind("127.0.0.1:0").unwrap();
+    let bad_addr = unused.local_addr().unwrap();
+    drop(unused);
+    let (task, _) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![target("good", good.addr), target("bad", bad_addr)],
+    )
+    .unwrap();
+    let outcome = runtime().block_on(task);
+    assert_eq!(outcome.total_target_outcomes, 2);
+    assert_eq!(outcome.successful_target_outcomes, 1);
+    assert_eq!(outcome.failed_target_outcomes, 1);
+    assert!(!probes(&good.finish()).is_empty());
+}
+
+#[test]
+fn pending_limit_failure_drains_reply_and_closes_session() {
+    let key = b"managed-send-failure-cleanup".to_vec();
+    let server = start_server(
+        ServerBehavior::DelayedEcho(Duration::from_millis(80)),
+        Some(key.clone()),
+    );
+    let mut managed = config(ManagedPacing::Staggered);
+    managed.client.duration = Some(Duration::from_millis(250));
+    managed.client.interval = Duration::from_millis(20);
+    managed.client.probe_timeout = Duration::from_secs(2);
+    managed.client.max_pending_probes = 1;
+    managed.final_drain = Duration::from_millis(50);
+    let mut configured = target("one", server.addr);
+    configured.auth = Some(ClientAuthConfig {
+        hmac_key: Some(key),
+    });
+    let (task, _) = ManagedClient::task(managed, vec![configured]).unwrap();
+    let outcome = runtime().block_on(async {
+        tokio::time::timeout(Duration::from_secs(1), task)
+            .await
+            .expect("send failure cleanup must shorten after draining the committed reply")
+    });
+    let records = server.finish();
+    let target = &outcome.recent_target_outcomes[0];
+    assert_eq!(outcome.failed_target_outcomes, 1);
+    assert!(matches!(
+        target.end_reason,
+        ManagedTargetEndReason::Failed(ManagedTargetFailure {
+            phase: ManagedTargetFailurePhase::Sending,
+            kind: ManagedTargetFailureKind::ResourceExhausted,
+            ..
+        })
+    ));
+    assert_eq!(target.cleanup_failure, None);
+    assert_eq!(target.packets_sent, 1);
+    assert_eq!(target.replies_received, 1);
+    assert_eq!(probes(&records).len(), 1);
+    assert!(has_close(&records));
+}
+
+#[test]
+fn event_loss_independence() {
+    let server = start_server(ServerBehavior::Echo, None);
+    let mut config = config(ManagedPacing::Staggered);
+    config.event_capacity = 1;
+    let (task, handle) = ManagedClient::task(config, vec![target("one", server.addr)]).unwrap();
+    let mut events = handle.subscribe().unwrap();
+    let outcome = runtime().block_on(task);
+    assert!(matches!(
+        events.try_recv(),
+        Err(broadcast::error::TryRecvError::Lagged(_))
+    ));
+    assert_eq!(outcome.successful_target_outcomes, 1);
+    assert_eq!(handle.status().lifecycle, ManagedLifecycle::Completed);
+    server.finish();
+}
+
+#[test]
+fn dropping_all_handles_does_not_stop() {
+    let server = start_server(ServerBehavior::Echo, None);
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![target("one", server.addr)],
+    )
+    .unwrap();
+    drop(handle);
+    let outcome = runtime().block_on(task);
+    assert_eq!(outcome.end_reason, ManagedEndReason::TargetsComplete);
+    server.finish();
+}
+
+#[test]
+fn task_drop_abandonment() {
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![target("one", "127.0.0.1:9".parse().unwrap())],
+    )
+    .unwrap();
+    let mut events = handle.subscribe().unwrap();
+    drop(task);
+    assert_eq!(handle.status().lifecycle, ManagedLifecycle::Abandoned);
+    assert_eq!(handle.status().total_target_outcomes, 0);
+    assert!(matches!(events.try_recv(), Ok(ManagedEvent::Abandoned)));
+}
+
+#[test]
+fn terminal_subscription_is_closed() {
+    let (task, handle) = ManagedClient::task(
+        config(ManagedPacing::Staggered),
+        vec![target("one", "127.0.0.1:9".parse().unwrap())],
+    )
+    .unwrap();
+    let mut task = Box::pin(task);
+    assert!(poll_once(task.as_mut()).is_ready());
+    assert!(matches!(
+        handle.subscribe(),
+        Err(ManagedSubscribeError::Closed)
+    ));
+}