diff --git a/Cargo.lock b/Cargo.lock
index 7e3280ae94..70a77fffed 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2771,6 +2771,7 @@ dependencies = [
  "ironrdp-rdpeai",
  "ironrdp-rdpei",
  "ironrdp-rdpel",
+ "ironrdp-rdpemt",
  "ironrdp-rdpeudp",
  "ironrdp-rdpeudp-tokio",
  "ironrdp-rdpewa",
diff --git a/crates/ironrdp-client/Cargo.toml b/crates/ironrdp-client/Cargo.toml
index d3b50eb465..d2c9a182ee 100644
--- a/crates/ironrdp-client/Cargo.toml
+++ b/crates/ironrdp-client/Cargo.toml
@@ -18,6 +18,9 @@ test = false
 [features]
 default = []
 
+# Internal integration-test access; not part of the supported public API.
+__test = []
+
 rustls = [
     "ironrdp-tls/rustls",
     "tokio-tungstenite/rustls-tls-native-roots",
@@ -50,6 +53,7 @@ webauthn = ["dep:ironrdp-rdpewa", "dep:ironrdp-rdpewa-native", "dvc-com-plugin"]
 vmconnect = ["dep:ironrdp-vmconnect"]
 location = ["dep:ironrdp-rdpel"]
 udp = [
+    "dep:ironrdp-rdpemt",
     "dep:ironrdp-rdpeudp",
     "dep:ironrdp-rdpeudp-tokio",
     "ironrdp-rdpeudp-tokio/rustls-aws-lc-rs",
@@ -82,6 +86,7 @@ ironrdp-displaycontrol = { path = "../ironrdp-displaycontrol", version = "0.8" }
 ironrdp-echo = { path = "../ironrdp-echo", version = "0.4" }
 ironrdp-egfx = { path = "../ironrdp-egfx", version = "0.3" }
 ironrdp-rdpei = { path = "../ironrdp-rdpei", version = "0.1" }
+ironrdp-rdpemt = { path = "../ironrdp-rdpemt", version = "0.1", optional = true }
 ironrdp-rdpeudp = { path = "../ironrdp-rdpeudp", version = "0.1", optional = true }
 ironrdp-rdpeudp-tokio = { path = "../ironrdp-rdpeudp-tokio", version = "0.1", optional = true }
 ironrdp-tls = { path = "../ironrdp-tls", version = "0.2" } # public
diff --git a/crates/ironrdp-client/src/lib.rs b/crates/ironrdp-client/src/lib.rs
index fe4b112898..0459c35972 100644
--- a/crates/ironrdp-client/src/lib.rs
+++ b/crates/ironrdp-client/src/lib.rs
@@ -16,3 +16,13 @@ pub mod rdp;
 mod clipboard;
 
 mod ws;
+
+#[cfg(all(feature = "udp", feature = "__test"))]
+#[doc(hidden)]
+pub mod udp;
+#[cfg(all(feature = "udp", not(feature = "__test")))]
+#[expect(
+    unreachable_pub,
+    reason = "the __test feature exposes this module to the shared integration tests"
+)]
+mod udp;
diff --git a/crates/ironrdp-client/src/rdp.rs b/crates/ironrdp-client/src/rdp.rs
index d184f13b11..676f085084 100644
--- a/crates/ironrdp-client/src/rdp.rs
+++ b/crates/ironrdp-client/src/rdp.rs
@@ -9,7 +9,7 @@ use std::io;
 use std::sync::Arc;
 #[cfg(feature = "location")]
 use std::sync::mpsc as std_mpsc;
-#[cfg(feature = "location")]
+#[cfg(any(feature = "location", feature = "udp"))]
 use std::time::Instant;
 
 #[cfg(feature = "clipboard")]
@@ -39,6 +39,7 @@ use ironrdp_pdu::input::mouse::PointerFlags;
     all(windows, feature = "webauthn")
 ))]
 use ironrdp_pdu::pdu_other_err;
+use ironrdp_pdu::rdp::autodetect::AutoDetectRequest;
 use ironrdp_pdu::rdp::multitransport::MultitransportResponsePdu;
 use ironrdp_pdu::rdp::session_info::ServerAutoReconnect;
 #[cfg(feature = "rdpdr")]
@@ -83,6 +84,8 @@ use ironrdp_rdpsnd_native::{RdpeaiCaptureBackend, cpal};
 
 use crate::config::{Config, RDCleanPathConfig, Transport};
 use crate::rail::{RailClient, RailControlEvent, RailEvent, RailInputEvent};
+#[cfg(feature = "udp")]
+use crate::udp::{disable_failed_tunnel, tunnel_auto_detect_requests, tunnel_auto_detect_sub_header};
 use ironrdp_rail::pdu::{ExecutePdu, ExecuteResultPdu};
 
 // ── Public event types ────────────────────────────────────────────────────────
@@ -3103,6 +3106,9 @@ async fn active_session(
     let mut graceful_shutdown_sent = false;
     let mut post_logon_redraw_requested = false;
     let mut pending_udp_payload: Option<Vec<u8>> = None;
+    // Auto-detect on the tunnel is timed against its own monotonic clock.
+    #[cfg(feature = "udp")]
+    let tunnel_clock = Instant::now();
     let mut initial_outputs = if *graceful_close_receiver.borrow_and_update() {
         graceful_shutdown_sent = true;
         Some(active_stage.graceful_shutdown()?)
@@ -3184,7 +3190,8 @@ async fn active_session(
                         Err(error) => return Err(ironrdp_session::custom_err!("read frame", error)),
                     };
                     trace!(?action, frame_length = payload.len(), "Frame received");
-                    let mut outputs = active_stage.process(&mut image, action, &payload)?;
+                    let mut outputs =
+                        active_stage.process_with_timestamp(&mut image, action, &payload, reader.last_read_at())?;
                     #[cfg(feature = "rdpdr")]
                     if let Some(output) = poll_deferred_rdpdr_output(&mut active_stage)? {
                         outputs.push(output);
@@ -3246,42 +3253,84 @@ async fn active_session(
                     }
                     ActiveSessionIteration::outputs(outputs)
                 }
-                udp_payload = async {
+                udp_message = async {
                     #[cfg(feature = "udp")]
                     {
                         match (udp_tunnel.transport.as_mut(), pending_udp_payload.is_none()) {
-                            (Some(transport), true) => transport.recv().await,
+                            (Some(transport), true) => transport
+                                .recv_message()
+                                .await
+                                .map(|message| (tunnel_auto_detect_requests(&message.sub_headers), message.data)),
                             (Some(_), false) | (None, _) => {
-                                core::future::pending::<Option<Vec<u8>>>().await
+                                core::future::pending::<Option<(Vec<AutoDetectRequest>, Vec<u8>)>>().await
                             }
                         }
                     }
                     #[cfg(not(feature = "udp"))]
                     {
-                        core::future::pending::<Option<Vec<u8>>>().await
+                        core::future::pending::<Option<(Vec<AutoDetectRequest>, Vec<u8>)>>().await
                     }
                 } => {
-                    match udp_payload {
+                    match udp_message {
                     None => {
-                        if active_stage.reliable_udp_dvc_tunnel_in_use() {
-                            return Ok(RdpControlFlow::TransportFailure(
-                                ironrdp_session::general_err!("reliable UDP tunnel closed"),
-                            ));
-                        }
                         #[cfg(feature = "udp")]
                         {
-                            udp_tunnel.transport = None;
+                            if let Err(error) = disable_failed_tunnel(
+                                &mut active_stage,
+                                &mut udp_tunnel.transport,
+                                ironrdp_session::general_err!("reliable UDP tunnel closed"),
+                            ) {
+                                return Ok(RdpControlFlow::TransportFailure(error));
+                            }
                         }
-                        active_stage.disable_reliable_udp_dvc_tunnel()?;
-                        warn!("Reliable UDP tunnel closed before Soft-Sync; continuing with TCP");
                         ActiveSessionIteration::outputs(Vec::new())
                     }
-                    Some(payload) if payload.is_empty() => {
-                        trace!("Ignoring reliable UDP tunnel PDU without higher-layer data");
-                        ActiveSessionIteration::outputs(Vec::new())
-                    }
-                    Some(payload) => {
-                        if active_stage.reliable_udp_dvc_tunnel_in_use() {
+                    Some((auto_detect_requests, payload)) => {
+                        #[cfg(feature = "udp")]
+                        {
+                            let received_at = ironrdp_core::MonotonicInstant::from_millis(
+                                u64::try_from(tunnel_clock.elapsed().as_millis()).unwrap_or(u64::MAX),
+                            );
+                            let responses = active_stage.process_tunnel_auto_detect(
+                                auto_detect_requests,
+                                payload.len(),
+                                received_at,
+                            );
+                            let sub_headers: Vec<_> = responses.iter().filter_map(tunnel_auto_detect_sub_header).collect();
+                            if !sub_headers.is_empty() && let Some(transport) = udp_tunnel.transport.as_ref() {
+                                let reply = ironrdp_rdpeudp_tokio::TunnelMessage {
+                                    sub_headers,
+                                    data: Vec::new(),
+                                };
+                                let Some(result) =
+                                    cancelable_operation(transport.send_message(reply), close_receiver).await
+                                else {
+                                    return Ok(RdpControlFlow::TerminatedGracefully(
+                                        GracefulDisconnectReason::UserInitiated,
+                                    ));
+                                };
+                                if let Err(error) = result {
+                                    if let Err(error) = disable_failed_tunnel(
+                                        &mut active_stage,
+                                        &mut udp_tunnel.transport,
+                                        ironrdp_session::custom_err!("answer reliable UDP tunnel auto-detect", error),
+                                    ) {
+                                        return Ok(RdpControlFlow::TransportFailure(error));
+                                    }
+                                    // This PDU arrived before any channel migrated. Once the
+                                    // tunnel fails, it cannot retain data for a future Soft-Sync.
+                                    pending_udp_payload = None;
+                                    continue;
+                                }
+                            }
+                        }
+                        #[cfg(not(feature = "udp"))]
+                        let _ = auto_detect_requests;
+
+                        if payload.is_empty() {
+                            trace!("Reliable UDP tunnel PDU without higher-layer data");
+                            ActiveSessionIteration::outputs(Vec::new())
+                        } else if active_stage.reliable_udp_dvc_tunnel_in_use() {
                             ActiveSessionIteration::tunnel(
                                 SoftSyncTunnelType::RELIABLE_UDP,
                                 active_stage.process_dvc_tunnel(
diff --git a/crates/ironrdp-client/src/udp.rs b/crates/ironrdp-client/src/udp.rs
new file mode 100644
index 0000000000..edef32363a
--- /dev/null
+++ b/crates/ironrdp-client/src/udp.rs
@@ -0,0 +1,63 @@
+//! Internal client handling for a reliable UDP tunnel.
+
+use ironrdp_core::{Decode, Encode};
+use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse};
+use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader};
+use ironrdp_rdpeudp_tokio::UdpTransport;
+use ironrdp_session::{ActiveStage, SessionError, SessionResult};
+use tracing::{debug, warn};
+
+/// Decodes the auto-detect requests among the sub-headers of a Tunnel Data PDU.
+///
+/// Each sub-header is the request structure itself: SubHeaderLength and SubHeaderType
+/// are headerLength and headerTypeId ([MS-RDPEMT] 2.2.1.1.1). Unknown requests are skipped.
+/// RTT requests are accepted for Windows interoperability in addition to the bandwidth
+/// requests listed by MS-RDPEMT; RTT encapsulation on a tunnel is not specified there.
+///
+/// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad
+pub fn tunnel_auto_detect_requests(sub_headers: &[TunnelSubHeader]) -> Vec<AutoDetectRequest> {
+    sub_headers
+        .iter()
+        .filter(|sub_header| sub_header.sub_header_type == SubHeaderType::AutoDetectRequest)
+        .filter_map(|sub_header| match redecode(sub_header) {
+            Ok(request) => Some(request),
+            Err(error) => {
+                debug!(%error, data = ?sub_header.data, "Ignoring an undecodable auto-detect request on the tunnel");
+                None
+            }
+        })
+        .collect()
+}
+
+/// Encodes a response as its wire-compatible tunnel sub-header.
+pub fn tunnel_auto_detect_sub_header(response: &AutoDetectResponse) -> Option<TunnelSubHeader> {
+    match redecode(response) {
+        Ok(sub_header) => Some(sub_header),
+        Err(error) => {
+            debug!(%error, ?response, "Could not encode an auto-detect response for the tunnel");
+            None
+        }
+    }
+}
+
+fn redecode<T: for<'de> Decode<'de>>(value: &dyn Encode) -> Result<T, String> {
+    let bytes = ironrdp_core::encode_vec(value).map_err(|error| error.to_string())?;
+    ironrdp_core::decode(&bytes).map_err(|error| error.to_string())
+}
+
+/// Drops a failed sideband and withdraws it from future Soft-Sync negotiation.
+/// A channel already moved to the tunnel cannot resume over TCP without reconnecting.
+/// Both receive closure and auto-detect send failures use this policy.
+pub fn disable_failed_tunnel(
+    stage: &mut ActiveStage,
+    transport: &mut Option<UdpTransport>,
+    error: SessionError,
+) -> SessionResult<()> {
+    if stage.reliable_udp_dvc_tunnel_in_use() {
+        return Err(error);
+    }
+    *transport = None;
+    stage.disable_reliable_udp_dvc_tunnel()?;
+    warn!(%error, "Reliable UDP tunnel failed before channel migration; continuing with TCP");
+    Ok(())
+}
diff --git a/crates/ironrdp-rdpeudp-tokio/src/error.rs b/crates/ironrdp-rdpeudp-tokio/src/error.rs
index 2ebe906487..2018af9bcc 100644
--- a/crates/ironrdp-rdpeudp-tokio/src/error.rs
+++ b/crates/ironrdp-rdpeudp-tokio/src/error.rs
@@ -114,6 +114,11 @@ pub enum UdpTransportErrorKind {
     /// A `send()` payload exceeds the wire `PayloadLength` field's 65535-byte
     /// capacity ([MS-RDPEMT] 2.2.2.3, `RDP_TUNNEL_DATA`).
     PayloadTooLarge { len: usize },
+
+    /// A `send_message()` message's sub-headers, `len` bytes encoded, do not
+    /// fit beside the 4-byte tunnel header in the one-byte `HeaderLength`
+    /// field ([MS-RDPEMT] 2.2.1.1).
+    SubHeadersTooLarge { len: usize },
 }
 
 impl fmt::Display for UdpTransportErrorKind {
@@ -140,6 +145,9 @@ impl fmt::Display for UdpTransportErrorKind {
                     "send payload of {len} bytes exceeds the 65535-byte tunnel data limit"
                 )
             }
+            Self::SubHeadersTooLarge { len } => {
+                write!(f, "{len} bytes of sub-headers exceed the 251 a tunnel header holds")
+            }
         }
     }
 }
@@ -157,7 +165,7 @@ impl core::error::Error for UdpTransportErrorKind {
             | Self::TunnelTimeout
             | Self::TunnelRejected { .. }
             | Self::DriverPanic => None,
-            Self::UnsupportedProtocol { .. } | Self::PayloadTooLarge { .. } => None,
+            Self::UnsupportedProtocol { .. } | Self::PayloadTooLarge { .. } | Self::SubHeadersTooLarge { .. } => None,
         }
     }
 }
diff --git a/crates/ironrdp-rdpeudp-tokio/src/framed.rs b/crates/ironrdp-rdpeudp-tokio/src/framed.rs
index f38f5ef361..ea74fbcd8f 100644
--- a/crates/ironrdp-rdpeudp-tokio/src/framed.rs
+++ b/crates/ironrdp-rdpeudp-tokio/src/framed.rs
@@ -33,9 +33,9 @@ impl FramedRead for UdpTransport {
             // puts no minimum on HigherLayerData, and [MS-RDPBCGR] 1.3.9 sends
             // the four Continuous Auto-Detection messages "encapsulated in the
             // RDP_TUNNEL_SUBHEADER structure ... over the sideband channels
-            // that are in active use". The tunnel has already taken what it
-            // needs from those subheaders by the time we get here, leaving a
-            // payload of nothing to pass on.
+            // that are in active use". `recv` drops those subheaders (a caller
+            // that needs them uses `recv_message`), leaving a payload of
+            // nothing to pass on.
             loop {
                 match self.recv().await {
                     Some(data) if data.is_empty() => {
@@ -74,11 +74,12 @@ mod tests {
     use tokio::sync::mpsc;
 
     use super::*;
+    use crate::transport::TunnelMessage;
 
     /// Build a `UdpTransport` backed by test channels (no real network).
-    fn test_transport() -> (UdpTransport, mpsc::Sender<Vec<u8>>, mpsc::Receiver<Vec<u8>>) {
-        let (incoming_tx, incoming_rx) = mpsc::channel::<Vec<u8>>(16);
-        let (outgoing_tx, outgoing_rx) = mpsc::channel::<Vec<u8>>(16);
+    fn test_transport() -> (UdpTransport, mpsc::Sender<TunnelMessage>, mpsc::Receiver<TunnelMessage>) {
+        let (incoming_tx, incoming_rx) = mpsc::channel::<TunnelMessage>(16);
+        let (outgoing_tx, outgoing_rx) = mpsc::channel::<TunnelMessage>(16);
 
         let transport = UdpTransport::from_channels(incoming_rx, outgoing_tx);
 
@@ -88,7 +89,7 @@ mod tests {
     #[tokio::test]
     async fn framed_read_delivers_one_message() {
         let (mut transport, feeder, _) = test_transport();
-        feeder.send(vec![0xDE, 0xAD, 0xBE, 0xEF]).await.unwrap();
+        feeder.send(vec![0xDE, 0xAD, 0xBE, 0xEF].into()).await.unwrap();
 
         let mut buf = BytesMut::new();
         let n = FramedRead::read(&mut transport, &mut buf).await.unwrap();
@@ -120,8 +121,8 @@ mod tests {
     async fn framed_read_does_not_mistake_an_empty_message_for_eof() {
         let (mut transport, feeder, _) = test_transport();
 
-        feeder.send(Vec::new()).await.unwrap();
-        feeder.send(vec![0x11, 0x22]).await.unwrap();
+        feeder.send(Vec::new().into()).await.unwrap();
+        feeder.send(vec![0x11, 0x22].into()).await.unwrap();
 
         let mut buf = BytesMut::new();
         let n = FramedRead::read(&mut transport, &mut buf).await.unwrap();
@@ -135,7 +136,7 @@ mod tests {
     async fn framed_read_still_reports_eof_after_an_empty_message() {
         let (mut transport, feeder, _) = test_transport();
 
-        feeder.send(Vec::new()).await.unwrap();
+        feeder.send(Vec::new().into()).await.unwrap();
         drop(feeder);
 
         let mut buf = BytesMut::new();
@@ -154,7 +155,7 @@ mod tests {
             .unwrap();
 
         let data = receiver.recv().await.unwrap();
-        assert_eq!(data, vec![0x01, 0x02, 0x03]);
+        assert_eq!(data, TunnelMessage::from(vec![0x01, 0x02, 0x03]));
     }
 
     #[tokio::test]
@@ -171,8 +172,8 @@ mod tests {
     async fn framed_read_multiple_messages_accumulate() {
         let (mut transport, feeder, _) = test_transport();
 
-        feeder.send(vec![0xAA, 0xBB]).await.unwrap();
-        feeder.send(vec![0xCC, 0xDD]).await.unwrap();
+        feeder.send(vec![0xAA, 0xBB].into()).await.unwrap();
+        feeder.send(vec![0xCC, 0xDD].into()).await.unwrap();
 
         let mut buf = BytesMut::new();
 
diff --git a/crates/ironrdp-rdpeudp-tokio/src/lib.rs b/crates/ironrdp-rdpeudp-tokio/src/lib.rs
index 517975ccdd..20fece5f1a 100644
--- a/crates/ironrdp-rdpeudp-tokio/src/lib.rs
+++ b/crates/ironrdp-rdpeudp-tokio/src/lib.rs
@@ -15,5 +15,6 @@ pub(crate) mod tunnel;
 pub use self::error::{DriverError, DriverErrorKind, UdpTransportError, UdpTransportErrorKind};
 pub use self::multitransport::MultitransportBootstrap;
 pub use self::transport::{
-    UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, UdpTransportSender, accept_udp, connect_udp,
+    TunnelMessage, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, UdpTransportSender, accept_udp,
+    connect_udp,
 };
diff --git a/crates/ironrdp-rdpeudp-tokio/src/transport.rs b/crates/ironrdp-rdpeudp-tokio/src/transport.rs
index d5b9c79605..0a2b86be15 100644
--- a/crates/ironrdp-rdpeudp-tokio/src/transport.rs
+++ b/crates/ironrdp-rdpeudp-tokio/src/transport.rs
@@ -29,7 +29,10 @@ use tokio::task::JoinHandle;
 use tracing::{debug, info, trace, warn};
 
 use crate::driver::Driver;
-use crate::error::{DriverError, DriverErrorExt as _, DriverErrorKind, UdpTransportError, UdpTransportErrorExt as _};
+use crate::error::{
+    DriverError, DriverErrorExt as _, DriverErrorKind, UdpTransportError, UdpTransportErrorExt as _,
+    UdpTransportErrorKind,
+};
 use crate::stream::{RdpeudpStream, SharedIo};
 use crate::tls::{tls_accept, tls_upgrade};
 use crate::tunnel::{read_tunnel_pdu, tunnel_data_loop, write_tunnel_pdu};
@@ -211,6 +214,28 @@ impl<T> Drop for AbortOnDrop<T> {
     }
 }
 
+/// One RDPEMT Tunnel Data PDU's content: the higher-layer data and the
+/// sub-headers carried beside it ([MS-RDPEMT] 2.2.2.3).
+///
+/// The sub-headers carry the Continuous Auto-Detection messages that
+/// [MS-RDPBCGR] 1.3.9 sends over a sideband channel in use, such as a
+/// bandwidth measurement's Start and Stop and the client's results. `data`
+/// may be empty when a message carries only sub-headers.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TunnelMessage {
+    pub sub_headers: Vec<ironrdp_rdpemt::TunnelSubHeader>,
+    pub data: Vec<u8>,
+}
+
+impl From<Vec<u8>> for TunnelMessage {
+    fn from(data: Vec<u8>) -> Self {
+        Self {
+            sub_headers: Vec::new(),
+            data,
+        }
+    }
+}
+
 /// A cloneable handle for sending data over an established UDP transport,
 /// obtained from [`UdpTransport::sender`].
 ///
@@ -218,7 +243,7 @@ impl<T> Drop for AbortOnDrop<T> {
 /// and receiving already run over separate channels fed by separate
 /// background tasks, so this never contends with a concurrent `recv()`.
 #[derive(Clone)]
-pub struct UdpTransportSender(mpsc::Sender<Vec<u8>>);
+pub struct UdpTransportSender(mpsc::Sender<TunnelMessage>);
 
 impl UdpTransportSender {
     /// Send a higher-layer data frame through the tunnel.
@@ -231,13 +256,39 @@ impl UdpTransportSender {
     /// Returns `PayloadTooLarge` if `data` exceeds 65535 bytes, the wire
     /// `PayloadLength` field's capacity ([MS-RDPEMT] 2.2.2.3).
     pub async fn send(&self, data: Vec<u8>) -> Result<(), UdpTransportError> {
-        if data.len() > usize::from(u16::MAX) {
-            debug!(len = data.len(), "Rejected oversized tunnel payload");
-            return Err(UdpTransportError::payload_too_large("send", data.len()));
+        self.send_message(TunnelMessage::from(data)).await
+    }
+
+    /// Send higher-layer data through the tunnel with sub-headers beside it
+    /// in the same Tunnel Data PDU.
+    ///
+    /// # Errors
+    ///
+    /// As [`Self::send`], for `message.data`, and `SubHeadersTooLarge` if
+    /// the sub-headers do not fit the one-byte `HeaderLength` field
+    /// ([MS-RDPEMT] 2.2.1.1), checked here for the same reason.
+    pub async fn send_message(&self, message: TunnelMessage) -> Result<(), UdpTransportError> {
+        if message.data.len() > usize::from(u16::MAX) {
+            debug!(len = message.data.len(), "Rejected oversized tunnel payload");
+            return Err(UdpTransportError::payload_too_large("send", message.data.len()));
+        }
+        let sub_headers_len = message
+            .sub_headers
+            .iter()
+            .map(ironrdp_rdpemt::TunnelSubHeader::wire_size)
+            .sum::<usize>();
+        // HeaderLength is one byte and counts the 4 byte fixed part of the tunnel
+        // header plus the sub-headers ([MS-RDPEMT] 2.2.1.1). The write pump's
+        // TunnelData encoder enforces the same bound; a test pins the two together.
+        if 4 /* RDP_TUNNEL_HEADER */ + sub_headers_len > usize::from(u8::MAX) {
+            return Err(UdpTransportError::new(
+                "send",
+                UdpTransportErrorKind::SubHeadersTooLarge { len: sub_headers_len },
+            ));
         }
 
         self.0
-            .send(data)
+            .send(message)
             .await
             .map_err(|_| UdpTransportError::driver("send", DriverError::connection_closed("send")))
     }
@@ -252,10 +303,10 @@ impl UdpTransportSender {
 /// Drop this handle to initiate shutdown of the background tasks.
 pub struct UdpTransport {
     /// Receives higher-layer data (DVC frames) from the tunnel.
-    data_rx: mpsc::Receiver<Vec<u8>>,
+    data_rx: mpsc::Receiver<TunnelMessage>,
 
     /// Sends higher-layer data into the tunnel for encryption and transmission.
-    data_tx: mpsc::Sender<Vec<u8>>,
+    data_tx: mpsc::Sender<TunnelMessage>,
 
     /// Shared I/O bridge between the driver and the TLS/RDPEMT layer.
     /// Held here so `shutdown()` can signal closure to both the
@@ -275,8 +326,16 @@ pub struct UdpTransport {
 impl UdpTransport {
     /// Receive the next higher-layer data frame from the tunnel.
     ///
-    /// Returns `None` when the tunnel is closed.
+    /// Returns `None` when the tunnel is closed. Any sub-headers that came
+    /// with the frame are dropped; use [`Self::recv_message`] to keep them.
     pub async fn recv(&mut self) -> Option<Vec<u8>> {
+        self.recv_message().await.map(|message| message.data)
+    }
+
+    /// Receive the next Tunnel Data PDU's content, sub-headers included.
+    ///
+    /// Returns `None` when the tunnel is closed.
+    pub async fn recv_message(&mut self) -> Option<TunnelMessage> {
         self.data_rx.recv().await
     }
 
@@ -296,6 +355,16 @@ impl UdpTransport {
         self.sender().send(data).await
     }
 
+    /// Send higher-layer data through the tunnel with sub-headers beside it
+    /// in the same Tunnel Data PDU.
+    ///
+    /// # Errors
+    ///
+    /// As [`UdpTransportSender::send_message`].
+    pub async fn send_message(&self, message: TunnelMessage) -> Result<(), UdpTransportError> {
+        self.sender().send_message(message).await
+    }
+
     /// Returns a cloneable handle for sending data, independent of this
     /// object's `&mut self`-requiring [`Self::recv`].
     ///
@@ -380,7 +449,7 @@ impl UdpTransport {
     /// For unit tests that exercise the channel-based API (FramedRead,
     /// FramedWrite) without needing a real UDP socket or TLS stack.
     #[cfg(test)]
-    pub(crate) fn from_channels(data_rx: mpsc::Receiver<Vec<u8>>, data_tx: mpsc::Sender<Vec<u8>>) -> Self {
+    pub(crate) fn from_channels(data_rx: mpsc::Receiver<TunnelMessage>, data_tx: mpsc::Sender<TunnelMessage>) -> Self {
         Self {
             data_rx,
             data_tx,
@@ -548,8 +617,8 @@ pub async fn connect_udp(config: UdpTransportConfig) -> Result<UdpTransport, Udp
     debug!("RDPEMT tunnel established, starting data pump");
 
     // Phase 5: Set up data channels and spawn the read pump
-    let (incoming_tx, incoming_rx) = mpsc::channel::<Vec<u8>>(64);
-    let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (incoming_tx, incoming_rx) = mpsc::channel::<TunnelMessage>(64);
+    let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<TunnelMessage>(64);
 
     // Read pump: TLS → RDPEMT decode → channel
     let pump_handle = AbortOnDrop::new(tokio::spawn(async move {
@@ -740,8 +809,8 @@ async fn accept_udp_inner(socket: UdpSocket, config: UdpAcceptConfig) -> Result<
     debug!("RDPEMT server tunnel established, starting data pump");
 
     // Phase 6: Set up data channels and spawn pumps (identical to client side)
-    let (incoming_tx, incoming_rx) = mpsc::channel::<Vec<u8>>(64);
-    let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (incoming_tx, incoming_rx) = mpsc::channel::<TunnelMessage>(64);
+    let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<TunnelMessage>(64);
 
     let pump_handle = AbortOnDrop::new(tokio::spawn(async move {
         tunnel_data_loop(&mut tls_read, &mut tunnel, &incoming_tx).await
@@ -818,15 +887,19 @@ where
 /// Shared by both `connect_udp` and `accept_udp`. Returns the first failure
 /// encountered rather than only logging it, so `shutdown()` can surface it
 /// to the caller instead of the pump silently going quiet.
-async fn write_pump<W>(tls_write: &mut W, outgoing_rx: &mut mpsc::Receiver<Vec<u8>>) -> Result<(), UdpTransportError>
+async fn write_pump<W>(
+    tls_write: &mut W,
+    outgoing_rx: &mut mpsc::Receiver<TunnelMessage>,
+) -> Result<(), UdpTransportError>
 where
     W: tokio::io::AsyncWrite + Unpin,
 {
-    while let Some(data) = outgoing_rx.recv().await {
-        let len = data.len();
+    while let Some(message) = outgoing_rx.recv().await {
+        let len = message.data.len();
+        let sub_headers = message.sub_headers.len();
         let pdu = ironrdp_rdpemt::TunnelData {
-            sub_headers: Vec::new(),
-            higher_layer_data: data,
+            sub_headers: message.sub_headers,
+            higher_layer_data: message.data,
         };
         let encoded = ironrdp_core::encode_vec(&pdu)
             .map_err(|error| UdpTransportError::rdpemt("write pump", ironrdp_rdpemt::RdpemtError::encode(error)))?;
@@ -838,7 +911,7 @@ where
             debug!(%error, "Write pump failed to flush tunnel data");
             UdpTransportError::tls("write pump", error)
         })?;
-        trace!(len, encoded_len = encoded.len(), "Sent tunnel data");
+        trace!(len, sub_headers, encoded_len = encoded.len(), "Sent tunnel data");
     }
     debug!("Write pump stopped, send channel closed");
     Ok(())
@@ -917,8 +990,8 @@ mod tests {
         static DRIVER_RUNNING: AtomicBool = AtomicBool::new(false);
         DRIVER_RUNNING.store(false, Ordering::SeqCst);
 
-        let (_incoming_tx, incoming_rx) = mpsc::channel::<Vec<u8>>(4);
-        let (outgoing_tx, _outgoing_rx) = mpsc::channel::<Vec<u8>>(4);
+        let (_incoming_tx, incoming_rx) = mpsc::channel::<TunnelMessage>(4);
+        let (outgoing_tx, _outgoing_rx) = mpsc::channel::<TunnelMessage>(4);
 
         let transport = UdpTransport {
             data_rx: incoming_rx,
@@ -960,7 +1033,7 @@ mod tests {
         let error = driver_exit_during_handshake("test", Ok(Err(driver_error)));
 
         assert!(
-            matches!(error.kind(), crate::error::UdpTransportErrorKind::Handshake(_)),
+            matches!(error.kind(), UdpTransportErrorKind::Handshake(_)),
             "got {error:?}, expected a Handshake error carrying the driver's own cause"
         );
     }
@@ -973,7 +1046,7 @@ mod tests {
         let error = driver_exit_during_handshake("test", Ok(Ok(())));
 
         assert!(
-            matches!(error.kind(), crate::error::UdpTransportErrorKind::Handshake(_)),
+            matches!(error.kind(), UdpTransportErrorKind::Handshake(_)),
             "got {error:?}, expected a Handshake error for an unexpectedly-clean driver exit"
         );
     }
@@ -988,7 +1061,7 @@ mod tests {
         let error = driver_exit_during_handshake("test", join_result);
 
         assert!(
-            matches!(error.kind(), crate::error::UdpTransportErrorKind::DriverPanic),
+            matches!(error.kind(), UdpTransportErrorKind::DriverPanic),
             "got {error:?}, expected DriverPanic"
         );
     }
@@ -1031,8 +1104,109 @@ mod tests {
         .expect("the driver branch should resolve almost immediately, well inside 5 seconds");
 
         assert!(
-            matches!(outcome.kind(), crate::error::UdpTransportErrorKind::Handshake(_)),
+            matches!(outcome.kind(), UdpTransportErrorKind::Handshake(_)),
             "got {outcome:?}, expected the driver's real error to surface"
         );
     }
+
+    /// A message's sub-headers go into the Tunnel Data PDU beside its data.
+    #[tokio::test]
+    async fn the_write_pump_encodes_sub_headers() {
+        use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader};
+        use tokio::io::AsyncReadExt as _;
+
+        let sub_header = TunnelSubHeader {
+            sub_header_type: SubHeaderType::AutoDetectRequest,
+            // A Bandwidth Measure Start: sequenceNumber 7, requestType 0x0014. The
+            // sub-header's own two bytes are its headerLength and headerTypeId.
+            data: vec![0x07, 0x00, 0x14, 0x00],
+        };
+        let (tx, mut rx) = mpsc::channel(4);
+        tx.send(TunnelMessage {
+            sub_headers: vec![sub_header.clone()],
+            data: vec![0xaa, 0xbb],
+        })
+        .await
+        .expect("queue message");
+        drop(tx);
+
+        let (mut writer, mut reader) = tokio::io::duplex(1024);
+        write_pump(&mut writer, &mut rx).await.expect("pump drains");
+        drop(writer);
+
+        let mut wire = Vec::new();
+        reader.read_to_end(&mut wire).await.expect("read back");
+        let pdu: TunnelData = ironrdp_core::decode(&wire).expect("decode");
+        assert_eq!(pdu.sub_headers, vec![sub_header]);
+        assert_eq!(pdu.higher_layer_data, vec![0xaa, 0xbb]);
+    }
+
+    /// `recv` keeps its old shape and drops sub-headers; `recv_message` keeps them.
+    #[tokio::test]
+    async fn recv_message_keeps_what_recv_drops() {
+        use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader};
+
+        let (incoming_tx, incoming_rx) = mpsc::channel(4);
+        let (outgoing_tx, _outgoing_rx) = mpsc::channel(4);
+        let mut transport = UdpTransport::from_channels(incoming_rx, outgoing_tx);
+        let message = TunnelMessage {
+            sub_headers: vec![TunnelSubHeader {
+                sub_header_type: SubHeaderType::AutoDetectResponse,
+                data: vec![0x01],
+            }],
+            data: vec![0x02],
+        };
+
+        incoming_tx.send(message.clone()).await.expect("queue");
+        incoming_tx.send(message.clone()).await.expect("queue");
+
+        assert_eq!(transport.recv().await, Some(vec![0x02]));
+        assert_eq!(transport.recv_message().await, Some(message));
+    }
+
+    /// Sub-headers that would overflow `HeaderLength` are refused at the
+    /// call, not left to fail the write pump; the largest that fit go out. The
+    /// boundary is also checked against the real `TunnelData` encoder, so the
+    /// precheck cannot drift from the bound the write pump enforces.
+    #[tokio::test]
+    async fn send_message_refuses_sub_headers_the_header_cannot_hold() {
+        use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader};
+
+        let (_incoming_tx, incoming_rx) = mpsc::channel(4);
+        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4);
+        let transport = UdpTransport::from_channels(incoming_rx, outgoing_tx);
+        let message = |data_len| TunnelMessage {
+            sub_headers: vec![TunnelSubHeader {
+                sub_header_type: SubHeaderType::AutoDetectRequest,
+                data: vec![0; data_len],
+            }],
+            data: Vec::new(),
+        };
+
+        // 4 (tunnel header) + 2 (sub-header header) + 249 = 255.
+        transport.send_message(message(249)).await.expect("fits");
+        let error = transport.send_message(message(250)).await.expect_err("one byte over");
+        assert!(matches!(
+            error.kind(),
+            UdpTransportErrorKind::SubHeadersTooLarge { len: 252 }
+        ));
+
+        assert_eq!(outgoing_rx.recv().await, Some(message(249)));
+        assert!(outgoing_rx.try_recv().is_err(), "the refused message was not queued");
+
+        let encode = |message: TunnelMessage| {
+            ironrdp_core::encode_vec(&TunnelData {
+                sub_headers: message.sub_headers,
+                higher_layer_data: message.data,
+            })
+        };
+        assert!(
+            encode(message(249)).is_ok(),
+            "the encoder accepts what the precheck accepts"
+        );
+        assert!(
+            encode(message(250)).is_err(),
+            "the encoder refuses what the precheck refuses"
+        );
+    }
 }
diff --git a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs
index 89c682bcbe..654c66b4ab 100644
--- a/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs
+++ b/crates/ironrdp-rdpeudp-tokio/src/tunnel.rs
@@ -15,6 +15,7 @@ use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
 use tracing::{debug, trace};
 
 use crate::error::{UdpTransportError, UdpTransportErrorExt as _};
+use crate::transport::TunnelMessage;
 
 /// Read a complete RDPEMT PDU from the stream using self-framing.
 ///
@@ -128,7 +129,7 @@ where
 pub(crate) async fn tunnel_data_loop<S>(
     stream: &mut S,
     tunnel: &mut RdpemtTunnel,
-    data_tx: &tokio::sync::mpsc::Sender<Vec<u8>>,
+    data_tx: &tokio::sync::mpsc::Sender<TunnelMessage>,
 ) -> Result<(), UdpTransportError>
 where
     S: AsyncRead + Unpin,
@@ -154,13 +155,16 @@ where
 
         while let Some(event) = tunnel.poll_event() {
             match event {
-                // `sub_headers` (e.g. auto-detect bandwidth measurement, MS-RDPBCGR
-                // 2.2.14) are not consumed here; this driver only wires the DVC
-                // payload through. A future auto-detect integration would need to
-                // dispatch them instead of discarding them.
-                TunnelEvent::Data { data, .. } => {
-                    trace!(len = data.len(), "Forwarding tunnel data");
-                    if data_tx.send(data).await.is_err() {
+                // Sub-headers go through with the data: they carry the
+                // auto-detect messages ([MS-RDPBCGR] 1.3.9) of a sideband
+                // channel in use, which the application answers.
+                TunnelEvent::Data { sub_headers, data } => {
+                    trace!(
+                        len = data.len(),
+                        sub_headers = sub_headers.len(),
+                        "Forwarding tunnel data"
+                    );
+                    if data_tx.send(TunnelMessage { sub_headers, data }).await.is_err() {
                         debug!("Tunnel data receiver dropped, stopping read pump");
                         // Application dropped the receiver
                         return Ok(());
@@ -263,4 +267,43 @@ mod tests {
         let mut cursor = io::Cursor::new(vec![0x02, 0x05, 0x00, 0x04, 0x48, 0x65]);
         assert!(read_tunnel_pdu(&mut cursor).await.is_err());
     }
+
+    /// Sub-headers reach the application with the data they came with. They
+    /// carry the auto-detect messages of a sideband channel in use
+    /// ([MS-RDPBCGR] 1.3.9), which used to be dropped here.
+    #[tokio::test]
+    async fn the_data_loop_forwards_sub_headers() {
+        use ironrdp_rdpemt::{SubHeaderType, TunnelConfig, TunnelCreateResponse, TunnelData, TunnelSubHeader};
+
+        let mut tunnel = RdpemtTunnel::client(TunnelConfig {
+            request_id: 1,
+            security_cookie: [0; 16],
+        });
+        let response = ironrdp_core::encode_vec(&TunnelCreateResponse {
+            hr_response: TunnelCreateResponse::S_OK,
+        })
+        .expect("encode response");
+        tunnel.handle_pdu(&response).expect("tunnel established");
+        while tunnel.poll_event().is_some() {}
+
+        let sub_header = TunnelSubHeader {
+            sub_header_type: SubHeaderType::AutoDetectResponse,
+            data: vec![0x0e, 0x00, 0x05, 0x00],
+        };
+        let wire = ironrdp_core::encode_vec(&TunnelData {
+            sub_headers: vec![sub_header.clone()],
+            higher_layer_data: Vec::new(),
+        })
+        .expect("encode data");
+
+        let (tx, mut rx) = tokio::sync::mpsc::channel(4);
+        let mut stream = io::Cursor::new(wire);
+        tunnel_data_loop(&mut stream, &mut tunnel, &tx)
+            .await
+            .expect("clean EOF");
+
+        let message = rx.recv().await.expect("one message");
+        assert_eq!(message.sub_headers, vec![sub_header]);
+        assert!(message.data.is_empty());
+    }
 }
diff --git a/crates/ironrdp-session/src/active_stage.rs b/crates/ironrdp-session/src/active_stage.rs
index 050111d751..1f028d8619 100644
--- a/crates/ironrdp-session/src/active_stage.rs
+++ b/crates/ironrdp-session/src/active_stage.rs
@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use ironrdp_bulk::{BulkCompressor, CompressionType as BulkCompressionType};
-use ironrdp_core::{ReadCursor, WriteBuf};
+use ironrdp_core::{MonotonicInstant, ReadCursor, WriteBuf};
 use ironrdp_displaycontrol::client::DisplayControlClient;
 use ironrdp_dvc::pdu::SoftSyncTunnelType;
 use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DvcMessageBatch, DynamicChannelMut, DynamicChannelRef};
@@ -10,7 +10,7 @@ use ironrdp_graphics::pointer::DecodedPointer;
 use ironrdp_pdu::gcc::{ChannelName, Monitor};
 use ironrdp_pdu::geometry::{ExclusiveRectangle, InclusiveRectangle, Rectangle as _};
 use ironrdp_pdu::input::fast_path::{FastPathInput, FastPathInputEvent};
-use ironrdp_pdu::rdp::autodetect::AutoDetectRequest;
+use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse};
 use ironrdp_pdu::rdp::capability_sets::WindowSupportLevel;
 use ironrdp_pdu::rdp::client_info::CompressionType;
 use ironrdp_pdu::rdp::headers::ShareDataPdu;
@@ -27,6 +27,7 @@ use ironrdp_rdpei::RdpeiClient;
 use ironrdp_svc::{StaticChannelSet, SvcMessage, SvcProcessor, SvcProcessorMessages};
 use tracing::{debug, warn};
 
+use crate::autodetect::AutoDetectResponder;
 use crate::fast_path::UpdateKind;
 use crate::image::DecodedImage;
 use crate::{SessionError, SessionErrorExt as _, SessionResult, fast_path, x224};
@@ -49,6 +50,8 @@ pub struct ActiveStage {
     window_support_level: Option<WindowSupportLevel>,
     graphics_output_needs_full_refresh: bool,
     damage_regions: Vec<InclusiveRectangle>,
+    /// Answers the auto-detect requests carried on the reliable UDP tunnel.
+    tunnel_auto_detect: AutoDetectResponder,
 }
 
 /// Builder for [`ActiveStage`].
@@ -107,6 +110,7 @@ impl ActiveStageBuilder {
             window_support_level: None,
             graphics_output_needs_full_refresh: false,
             damage_regions: Vec::new(),
+            tunnel_auto_detect: AutoDetectResponder::default(),
         }
     }
 }
@@ -183,15 +187,37 @@ impl ActiveStage {
     }
 
     /// Process a frame received from the server.
+    ///
+    /// Without an arrival time, bandwidth measurements are answered with an untimed,
+    /// zero-byte result; see [`Self::process_with_timestamp`].
     pub fn process(
         &mut self,
         image: &mut DecodedImage,
         action: Action,
         frame: &[u8],
+    ) -> SessionResult<Vec<ActiveStageOutput>> {
+        self.process_with_timestamp(image, action, frame, None)
+    }
+
+    /// Process a frame received from the server, together with its arrival time.
+    ///
+    /// The clock stays outside this state machine: `received_at` is the time the transport
+    /// read the frame, from the same monotonic clock for every frame. Frames that were
+    /// buffered must keep their read time so a bandwidth measurement reflects network arrival
+    /// rather than the time spent decoding earlier frames.
+    pub fn process_with_timestamp(
+        &mut self,
+        image: &mut DecodedImage,
+        action: Action,
+        frame: &[u8],
+        received_at: Option<MonotonicInstant>,
     ) -> SessionResult<Vec<ActiveStageOutput>> {
         self.damage_regions.clear();
         let (mut stage_outputs, processor_updates) = match action {
             Action::FastPath => {
+                // TLS-protected fast-path frames have no RDP Security Header, so the
+                // continuous bandwidth count includes the entire frame.
+                self.x224_processor.record_bandwidth_bytes(frame.len());
                 let mut output = WriteBuf::new();
                 let processor_updates =
                     self.fast_path_processor
@@ -202,7 +228,9 @@ impl ActiveStage {
                 )
             }
             Action::X224 => {
-                let x224_outputs = self.x224_processor.process(frame, &mut self.bulk_decompressor)?;
+                let x224_outputs =
+                    self.x224_processor
+                        .process_with_timestamp(frame, &mut self.bulk_decompressor, received_at)?;
                 let mut stage_outputs = Vec::new();
                 let mut processor_updates = Vec::new();
 
@@ -562,6 +590,33 @@ impl ActiveStage {
             .is_some_and(|drdynvc| drdynvc.switched_to_tunnel(SoftSyncTunnelType::RELIABLE_UDP))
     }
 
+    /// Answers the auto-detect requests the server sends in the sub-headers of one Tunnel Data
+    /// PDU ([MS-RDPEMT] 2.2.1.1.1), and counts the `data_len` bytes of higher-layer data the PDU
+    /// carries for an open bandwidth measurement. Returns the responses to send back on the
+    /// tunnel.
+    ///
+    /// The tunnel keeps its own measurement, apart from the message channel's, because a
+    /// measurement on the tunnel counts only the data that follows the tunnel PDU header
+    /// ([MS-RDPBCGR] 3.2.5.14). As on the message channel, received bytes are counted before
+    /// handling control messages: a Start resets the count, and a Stop includes the carrying
+    /// PDU's data in the result. Sub-header bytes themselves never count on the tunnel.
+    /// `received_at` is the time the PDU arrived, from one monotonic clock for the whole tunnel.
+    ///
+    /// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad
+    /// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6
+    pub fn process_tunnel_auto_detect(
+        &mut self,
+        requests: Vec<AutoDetectRequest>,
+        data_len: usize,
+        received_at: MonotonicInstant,
+    ) -> Vec<AutoDetectResponse> {
+        self.tunnel_auto_detect.record_bytes(data_len);
+        requests
+            .into_iter()
+            .filter_map(|request| self.tunnel_auto_detect.respond(request, Some(received_at)))
+            .collect()
+    }
+
     /// Returns the Soft-Sync tunnel selected for client messages on `channel_id`.
     pub fn dvc_tunnel_for_channel(&self, channel_id: u32) -> Option<SoftSyncTunnelType> {
         self.x224_processor
diff --git a/crates/ironrdp-session/src/autodetect.rs b/crates/ironrdp-session/src/autodetect.rs
new file mode 100644
index 0000000000..8c43f8df31
--- /dev/null
+++ b/crates/ironrdp-session/src/autodetect.rs
@@ -0,0 +1,148 @@
+//! Client answers to the network auto-detect requests of [MS-RDPBCGR] 2.2.14.
+//!
+//! [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/dc672839-4f4e-40b1-a71c-cd6a959baa38
+
+use ironrdp_core::MonotonicInstant;
+use ironrdp_pdu::rdp::autodetect::{
+    AutoDetectRequest, AutoDetectResponse, BW_RESULTS_CONNECT_TIME, BW_RESULTS_CONTINUOUS, BW_START_CONNECT_TIME,
+    BW_START_RELIABLE_UDP, BW_STOP_CONNECT_TIME, BW_STOP_RELIABLE_UDP,
+};
+use tracing::debug;
+
+/// Answers the RTT and bandwidth measurement requests that arrive on one transport.
+///
+/// Each transport keeps its own, because a continuous measurement counts the data received on
+/// the transport it runs on ([MS-RDPBCGR] 3.2.5.14). Timestamps come from the caller, from the
+/// same monotonic clock for every request on that transport.
+///
+/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6
+#[derive(Debug, Default)]
+pub(crate) struct AutoDetectResponder {
+    bandwidth: Option<BandwidthMeasurement>,
+}
+
+/// Size of the auto-detect header fields [MS-RDPBCGR] 3.2.5.14 counts along with payloadLength:
+/// headerLength, headerTypeId, sequenceNumber, requestType and payloadLength itself.
+///
+/// [MS-RDPBCGR]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/16ffa852-8aa7-481c-99a0-36c1a9a198f6
+const AUTO_DETECT_HEADER_LEN: u32 = 8;
+
+/// Reported as `timeDelta` for a window that was not timed, and the floor for one that was:
+/// a server computing `byteCount * 8 / timeDelta` divides by it ([MS-RDPBCGR] 3.3.5.14).
+const UNMEASURABLE_INTERVAL_MS: u32 = 1;
+
+/// Bytes a connect-time Payload or Stop adds to the count: payloadLength plus the auto-detect
+/// header, but not the security header. This is the rule the connector applies to the same PDUs.
+fn counted_len(payload_len: usize) -> u32 {
+    u32::try_from(payload_len)
+        .unwrap_or(u32::MAX)
+        .saturating_add(AUTO_DETECT_HEADER_LEN)
+}
+
+#[derive(Debug)]
+struct BandwidthMeasurement {
+    started_at: MonotonicInstant,
+    bytes: u32,
+    continuous: bool,
+}
+
+impl AutoDetectResponder {
+    /// Counts received bytes while a continuous bandwidth window is open.
+    pub(crate) fn record_bytes(&mut self, bytes: usize) {
+        if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| measurement.continuous) {
+            measurement.bytes = measurement
+                .bytes
+                .saturating_add(u32::try_from(bytes).unwrap_or(u32::MAX));
+        }
+    }
+
+    /// Returns the response `request` calls for, if any.
+    ///
+    /// Without an arrival time, a bandwidth measurement is answered with an untimed, zero-byte
+    /// result.
+    pub(crate) fn respond(
+        &mut self,
+        request: AutoDetectRequest,
+        received_at: Option<MonotonicInstant>,
+    ) -> Option<AutoDetectResponse> {
+        match request {
+            AutoDetectRequest::RttRequest { sequence_number, .. } => {
+                Some(AutoDetectResponse::RttResponse { sequence_number })
+            }
+            AutoDetectRequest::BandwidthMeasureStart { request_type, .. }
+                if matches!(request_type, BW_START_CONNECT_TIME | BW_START_RELIABLE_UDP) =>
+            {
+                self.bandwidth = received_at.map(|started_at| BandwidthMeasurement {
+                    started_at,
+                    bytes: 0,
+                    continuous: request_type == BW_START_RELIABLE_UDP,
+                });
+                None
+            }
+            AutoDetectRequest::BandwidthMeasurePayload { payload, .. } => {
+                if let Some(measurement) = self.bandwidth.as_mut().filter(|measurement| !measurement.continuous) {
+                    measurement.bytes = measurement.bytes.saturating_add(counted_len(payload.len()));
+                }
+                None
+            }
+            AutoDetectRequest::BandwidthMeasureStop {
+                sequence_number,
+                request_type,
+                payload,
+            } if matches!(request_type, BW_STOP_CONNECT_TIME | BW_STOP_RELIABLE_UDP) => {
+                let continuous = request_type == BW_STOP_RELIABLE_UDP;
+                let measurement = self
+                    .bandwidth
+                    .take()
+                    .filter(|measurement| measurement.continuous == continuous);
+                let stop_bytes = if continuous {
+                    0
+                } else {
+                    counted_len(payload.as_ref().map_or(0, Vec::len))
+                };
+                let (time_delta_ms, byte_count) = match (measurement, received_at) {
+                    (Some(measurement), Some(stopped_at)) => (
+                        u32::try_from(stopped_at.duration_since(measurement.started_at).as_millis())
+                            .unwrap_or(u32::MAX)
+                            .max(UNMEASURABLE_INTERVAL_MS),
+                        measurement.bytes.saturating_add(stop_bytes),
+                    ),
+                    (Some(measurement), None) => {
+                        // The window was timed but this Stop was not, so there is nothing to
+                        // divide the count by. Log the drop so it does not look like the
+                        // ordinary no-window case.
+                        debug!(
+                            dropped_bytes = measurement.bytes,
+                            "Bandwidth Measure Stop arrived with no arrival time although its window was open; \
+                             dropping the accumulated count"
+                        );
+                        (UNMEASURABLE_INTERVAL_MS, 0)
+                    }
+                    (None, _) => (UNMEASURABLE_INTERVAL_MS, 0),
+                };
+                Some(AutoDetectResponse::BandwidthMeasureResults {
+                    sequence_number,
+                    response_type: if continuous {
+                        BW_RESULTS_CONTINUOUS
+                    } else {
+                        BW_RESULTS_CONNECT_TIME
+                    },
+                    time_delta_ms,
+                    byte_count,
+                })
+            }
+            request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => {
+                // The TCP message-channel processor surfaces this request itself. Keep this
+                // arm for the UDP tunnel responder introduced in PR #2009, which handles
+                // auto-detect requests without passing through that processor.
+                debug!(?request, "Received network characteristics from server");
+                None
+            }
+            request => {
+                // Measurements for a lossy transport are not answered here.
+                debug!(?request, "Ignoring auto-detect request for another transport");
+                None
+            }
+        }
+    }
+}
diff --git a/crates/ironrdp-session/src/lib.rs b/crates/ironrdp-session/src/lib.rs
index 3103da0a37..84420484ed 100644
--- a/crates/ironrdp-session/src/lib.rs
+++ b/crates/ironrdp-session/src/lib.rs
@@ -11,6 +11,7 @@ pub mod rfx; // FIXME: maybe this module should not be in this crate
 pub mod x224;
 
 mod active_stage;
+mod autodetect;
 mod palette;
 
 use core::fmt;
diff --git a/crates/ironrdp-session/src/x224/mod.rs b/crates/ironrdp-session/src/x224/mod.rs
index d006393ba4..7e64c7ebf8 100644
--- a/crates/ironrdp-session/src/x224/mod.rs
+++ b/crates/ironrdp-session/src/x224/mod.rs
@@ -1,9 +1,9 @@
 use ironrdp_bulk::BulkCompressor;
-use ironrdp_core::{Decode as _, ReadCursor, WriteBuf, decode};
+use ironrdp_core::{Decode as _, MonotonicInstant, ReadCursor, WriteBuf, decode};
 use ironrdp_dvc::{DrdynvcClient, DvcClientProcessor, DynamicChannelMut, DynamicChannelRef};
 use ironrdp_pdu::gcc::{ChannelName, Monitor};
 use ironrdp_pdu::mcs::{DisconnectProviderUltimatum, DisconnectReason, McsMessage, SendDataIndicationCtx};
-use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu};
+use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectRspPdu};
 use ironrdp_pdu::rdp::client_info::CompressionType;
 use ironrdp_pdu::rdp::headers::{
     BasicSecurityHeader, BasicSecurityHeaderFlags, CompressionFlags, IoChannelPdu, ShareDataCtx, ShareDataPdu,
@@ -18,6 +18,7 @@ use ironrdp_svc::{
 };
 use tracing::debug;
 
+use crate::autodetect::AutoDetectResponder;
 use crate::{SessionError, SessionErrorExt as _, SessionResult, reason_err};
 
 /// X224 Processor output
@@ -72,7 +73,7 @@ pub enum ProcessorOutput {
     /// Auto-detect network characteristics from server ([\[MS-RDPBCGR\] 2.2.14]).
     ///
     /// Currently only surfaces [`AutoDetectRequest::NetworkCharacteristicsResult`].
-    /// RTT requests are handled internally with automatic responses.
+    /// RTT and bandwidth measurement requests are answered internally.
     ///
     /// [\[MS-RDPBCGR\] 2.2.14]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpbcgr/dc672839-4f4e-40b1-a71c-cd6a959baa38
     AutoDetect(AutoDetectRequest),
@@ -107,6 +108,7 @@ pub struct Processor {
     io_channel_id: u16,
     message_channel_id: Option<u16>,
     share_id: u32,
+    auto_detect: AutoDetectResponder,
 }
 
 impl Processor {
@@ -123,6 +125,7 @@ impl Processor {
             io_channel_id,
             message_channel_id,
             share_id,
+            auto_detect: AutoDetectResponder::default(),
         }
     }
 
@@ -212,6 +215,19 @@ impl Processor {
         &mut self,
         frame: &[u8],
         bulk_decompressor: &mut Option<BulkCompressor>,
+    ) -> SessionResult<Vec<ProcessorOutput>> {
+        self.process_with_timestamp(frame, bulk_decompressor, None)
+    }
+
+    /// Processes a frame with its driver-observed arrival time.
+    ///
+    /// Supply the same monotonic clock for every frame. Without a timestamp,
+    /// bandwidth replies use a conservative zero-byte measurement.
+    pub fn process_with_timestamp(
+        &mut self,
+        frame: &[u8],
+        bulk_decompressor: &mut Option<BulkCompressor>,
+        received_at: Option<MonotonicInstant>,
     ) -> SessionResult<Vec<ProcessorOutput>> {
         let data_ctx: SendDataIndicationCtx<'_> = match ironrdp_pdu::mcs::decode_send_data_indication(frame) {
             Ok(data_ctx) => data_ctx,
@@ -233,10 +249,12 @@ impl Processor {
         let channel_id = data_ctx.channel_id;
 
         if channel_id == self.io_channel_id {
-            self.process_io_channel_data_indication(data_ctx, bulk_decompressor)
+            self.process_io_channel_data_indication(data_ctx, frame.len(), bulk_decompressor)
         } else if self.message_channel_id == Some(channel_id) {
-            self.process_message_channel(data_ctx)
+            self.process_message_channel(data_ctx, received_at)
         } else {
+            // TLS-protected SVC data has no RDP Security Header: include TPKT/X224/MCS.
+            self.record_bandwidth_bytes(frame.len());
             let maximum_chunk_size = self.static_channels.maximum_chunk_size();
             if let Some(svc) = self.static_channels.get_by_channel_id_mut(channel_id) {
                 let response_pdus = svc.process(data_ctx.user_data).map_err(SessionError::pdu)?;
@@ -251,6 +269,7 @@ impl Processor {
     fn process_io_channel_data_indication(
         &mut self,
         data_ctx: SendDataIndicationCtx<'_>,
+        frame_len: usize,
         bulk_decompressor: &mut Option<BulkCompressor>,
     ) -> SessionResult<Vec<ProcessorOutput>> {
         debug_assert_eq!(data_ctx.channel_id, self.io_channel_id);
@@ -261,9 +280,13 @@ impl Processor {
             ironrdp_pdu::rdp::headers::decode_io_channel(data_ctx),
             Ok(IoChannelPdu::MultitransportRequest(_))
         ) {
+            self.record_bandwidth_bytes(data_ctx.user_data.len() - BasicSecurityHeader::FIXED_PART_SIZE);
             return self.process_io_channel(data_ctx, bulk_decompressor);
         }
 
+        // Ordinary TLS-protected IO data has no RDP Security Header. Count the
+        // entire frame once, including framing and any concatenated Share Control PDUs.
+        self.record_bandwidth_bytes(frame_len);
         let mut outputs = Vec::new();
         let mut offset = 0usize;
         let data = data_ctx.user_data;
@@ -445,6 +468,13 @@ impl Processor {
         Ok(decompressed)
     }
 
+    /// Counts received session bytes while a continuous bandwidth window is open.
+    /// Exclude framing only when an RDP Security Header is present, counting just
+    /// the bytes after that header ([MS-RDPBCGR] 3.2.5.14).
+    pub(crate) fn record_bandwidth_bytes(&mut self, bytes: usize) {
+        self.auto_detect.record_bytes(bytes);
+    }
+
     /// Process a PDU received on the MCS message channel: auto-detect
     /// ([MS-RDPBCGR] 2.2.14), multitransport ([MS-RDPBCGR] 2.2.15), or
     /// Heartbeat ([MS-RDPBCGR] 2.2.16.1).
@@ -457,13 +487,18 @@ impl Processor {
     /// session-fatal decode error: this channel is forward-safe for future
     /// message-channel PDU types the same way the connect-time demux
     /// (`ironrdp-connector`) already is.
-    fn process_message_channel(&self, data_ctx: SendDataIndicationCtx<'_>) -> SessionResult<Vec<ProcessorOutput>> {
+    fn process_message_channel(
+        &mut self,
+        data_ctx: SendDataIndicationCtx<'_>,
+        received_at: Option<MonotonicInstant>,
+    ) -> SessionResult<Vec<ProcessorOutput>> {
         let Some(message_channel_id) = self.message_channel_id else {
             return Err(reason_err!("message channel", "no message channel negotiated"));
         };
 
         let mut peek = ReadCursor::new(data_ctx.user_data);
         let security_header = BasicSecurityHeader::decode(&mut peek).map_err(SessionError::decode)?;
+        self.record_bandwidth_bytes(peek.len());
         let flags = security_header
             .flags
             .difference(BasicSecurityHeaderFlags::RESET_SEQNO | BasicSecurityHeaderFlags::IGNORE_SEQNO);
@@ -495,29 +530,23 @@ impl Processor {
 
         let req = decode::<AutoDetectReqPdu>(data_ctx.user_data).map_err(SessionError::decode)?;
 
-        match req.request {
-            AutoDetectRequest::RttRequest { sequence_number, .. } => {
-                let response = AutoDetectRspPdu::new(AutoDetectResponse::RttResponse { sequence_number });
-                let mut frame = WriteBuf::new();
-                ironrdp_pdu::mcs::encode_send_data_request(
-                    self.user_channel_id,
-                    message_channel_id,
-                    &response,
-                    &mut frame,
-                )
-                .map_err(SessionError::encode)?;
-                debug!(sequence_number, "Responded to auto-detect RTT request");
-                Ok(vec![ProcessorOutput::ResponseFrame(frame.into_inner())])
-            }
-            req @ AutoDetectRequest::NetworkCharacteristicsResult { .. } => {
-                debug!(?req, "Received network characteristics from server");
-                Ok(vec![ProcessorOutput::AutoDetect(req)])
-            }
-            req => {
-                debug!(?req, "Auto-detect request not yet implemented");
-                Ok(Vec::new())
-            }
+        if let request @ AutoDetectRequest::NetworkCharacteristicsResult { .. } = req.request {
+            debug!(?request, "Received network characteristics from server");
+            return Ok(vec![ProcessorOutput::AutoDetect(request)]);
         }
+        let Some(response) = self.auto_detect.respond(req.request, received_at) else {
+            return Ok(Vec::new());
+        };
+        debug!(?response, "Responding to an auto-detect request");
+        let mut frame = WriteBuf::new();
+        ironrdp_pdu::mcs::encode_send_data_request(
+            self.user_channel_id,
+            message_channel_id,
+            &AutoDetectRspPdu::new(response),
+            &mut frame,
+        )
+        .map_err(SessionError::encode)?;
+        Ok(vec![ProcessorOutput::ResponseFrame(frame.into_inner())])
     }
 
     /// Encodes an Initiate Multitransport Response on the MCS message channel.
@@ -638,14 +667,17 @@ mod tests {
     fn processor_surfaces_multitransport_request_on_message_channel() {
         let request = multitransport_request();
         let encoded = encode_vec(&request).expect("encode multitransport request");
-        let processor = Processor::new(StaticChannelSet::new(), 1002, 1003, Some(1004), 0);
+        let mut processor = Processor::new(StaticChannelSet::new(), 1002, 1003, Some(1004), 0);
 
         let outputs = processor
-            .process_message_channel(SendDataIndicationCtx {
-                initiator_id: 1002,
-                channel_id: 1004,
-                user_data: &encoded,
-            })
+            .process_message_channel(
+                SendDataIndicationCtx {
+                    initiator_id: 1002,
+                    channel_id: 1004,
+                    user_data: &encoded,
+                },
+                None,
+            )
             .expect("surface multitransport request");
 
         assert!(matches!(
@@ -667,6 +699,7 @@ mod tests {
                     channel_id: 1003,
                     user_data: &encoded,
                 },
+                encoded.len(),
                 &mut None,
             )
             .expect("ignore a misrouted optional multitransport request");
@@ -884,6 +917,7 @@ mod tests {
                     channel_id: 1003,
                     user_data: &user_data,
                 },
+                user_data.len(),
                 &mut None,
             )
             .expect("concatenated Share Control PDUs should be split");
@@ -905,6 +939,7 @@ mod tests {
                     channel_id: 1003,
                     user_data: &[0x06],
                 },
+                1,
                 &mut None,
             )
             .expect_err("a truncated totalLength field is invalid");
@@ -925,6 +960,7 @@ mod tests {
                     channel_id: 1003,
                     user_data: &user_data,
                 },
+                user_data.len(),
                 &mut None,
             )
             .expect_err("an overrunning concatenated totalLength is invalid");
@@ -950,6 +986,7 @@ mod tests {
                     channel_id: 1003,
                     user_data: &user_data,
                 },
+                user_data.len(),
                 &mut None,
             )
             .expect("invalid first totalLength should fall back to whole-buffer decode");
diff --git a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs
index 4591f3fe92..302e0981f9 100644
--- a/crates/ironrdp-testsuite-core/tests/session/autodetect.rs
+++ b/crates/ironrdp-testsuite-core/tests/session/autodetect.rs
@@ -1,10 +1,22 @@
 use std::borrow::Cow;
 
 use ironrdp_core::encode_vec;
+use ironrdp_dvc::DrdynvcClient;
+use ironrdp_dvc::pdu::{CapabilitiesRequestPdu, CapsVersion, DrdynvcServerPdu};
+use ironrdp_graphics::image_processing::PixelFormat;
+use ironrdp_pdu::Action;
+use ironrdp_pdu::fast_path::{EncryptionFlags, FastPathHeader, FastPathUpdatePdu, Fragmentation, UpdateCode};
 use ironrdp_pdu::mcs::{McsMessage, SendDataIndication};
 use ironrdp_pdu::rdp::autodetect::{AutoDetectReqPdu, AutoDetectRequest, AutoDetectResponse, AutoDetectRspPdu};
+use ironrdp_pdu::rdp::headers::{
+    BasicSecurityHeader, BasicSecurityHeaderFlags, ServerDeactivateAll, ShareControlHeader, ShareControlPdu,
+};
+use ironrdp_pdu::rdp::multitransport::{MultitransportRequestPdu, RequestedProtocol};
+use ironrdp_pdu::rdp::vc::{ChannelControlFlags, ChannelPduHeader};
 use ironrdp_pdu::x224::X224;
+use ironrdp_session::image::DecodedImage;
 use ironrdp_session::x224::Processor;
+use ironrdp_session::{ActiveStage, ActiveStageBuilder, ActiveStageOutput};
 use ironrdp_svc::StaticChannelSet;
 
 const USER_CHANNEL_ID: u16 = 1002;
@@ -33,10 +45,13 @@ fn process_frame(processor: &mut Processor, frame: &[u8]) -> Vec<ironrdp_session
 fn encode_server_autodetect(request: AutoDetectRequest) -> Vec<u8> {
     let pdu = AutoDetectReqPdu::new(request);
     let user_data = encode_vec(&pdu).unwrap();
+    encode_server_data(MESSAGE_CHANNEL_ID, user_data)
+}
 
+fn encode_server_data(channel_id: u16, user_data: Vec<u8>) -> Vec<u8> {
     let indication = McsMessage::SendDataIndication(SendDataIndication {
         initiator_id: USER_CHANNEL_ID,
-        channel_id: MESSAGE_CHANNEL_ID,
+        channel_id,
         user_data: Cow::Owned(user_data),
     });
 
@@ -130,7 +145,7 @@ fn bandwidth_measure_stop_does_not_crash() {
     let frame = encode_server_autodetect(request);
 
     let outputs = process_frame(&mut processor, &frame);
-    assert!(outputs.is_empty(), "BW stop should produce no output");
+    assert_eq!(bandwidth_result(&outputs), (200, 0x000b, 1, 0));
 }
 
 #[test]
@@ -142,3 +157,381 @@ fn bandwidth_measure_payload_does_not_crash() {
     let outputs = process_frame(&mut processor, &frame);
     assert!(outputs.is_empty(), "BW payload should produce no output");
 }
+
+fn timed_request(
+    processor: &mut Processor,
+    request: AutoDetectRequest,
+    millis: u64,
+) -> Vec<ironrdp_session::x224::ProcessorOutput> {
+    processor
+        .process_with_timestamp(
+            &encode_server_autodetect(request),
+            &mut None,
+            Some(ironrdp_core::MonotonicInstant::from_millis(millis)),
+        )
+        .expect("process timed auto-detect request")
+}
+
+fn bandwidth_result(outputs: &[ironrdp_session::x224::ProcessorOutput]) -> (u16, u16, u32, u32) {
+    let [ironrdp_session::x224::ProcessorOutput::ResponseFrame(frame)] = outputs else {
+        panic!("expected exactly one bandwidth response");
+    };
+    bandwidth_result_frame(frame)
+}
+
+fn bandwidth_result_frame(frame: &[u8]) -> (u16, u16, u32, u32) {
+    let X224(McsMessage::SendDataRequest(message)) = ironrdp_core::decode::<X224<McsMessage<'_>>>(frame).unwrap()
+    else {
+        panic!("expected main-channel response");
+    };
+    assert_eq!(message.channel_id, MESSAGE_CHANNEL_ID);
+    let response = ironrdp_core::decode::<AutoDetectRspPdu>(&message.user_data).unwrap();
+    let AutoDetectResponse::BandwidthMeasureResults {
+        sequence_number,
+        response_type,
+        time_delta_ms,
+        byte_count,
+    } = response.response
+    else {
+        panic!("expected bandwidth results");
+    };
+    (sequence_number, response_type, time_delta_ms, byte_count)
+}
+
+#[test]
+fn continuous_measurement_excludes_message_channel_security_headers() {
+    let mut processor = make_processor();
+    assert!(timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10).is_empty());
+    timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 20);
+    let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(3), 35);
+    // RTT and Stop have six-byte auto-detect headers. Neither four-byte
+    // security header, nor TPKT/X224/MCS framing, belongs to the count.
+    assert_eq!(bandwidth_result(&result), (3, 0x000b, 25, 12));
+}
+
+#[test]
+fn continuous_measurement_counts_io_framing_once_for_concatenated_pdus() {
+    let mut processor = make_processor();
+    let pdu = encode_vec(&ShareControlHeader {
+        share_control_pdu: ShareControlPdu::ServerDeactivateAll(ServerDeactivateAll),
+        pdu_source: USER_CHANNEL_ID,
+        share_id: SHARE_ID,
+    })
+    .unwrap();
+    let frame = encode_server_data(IO_CHANNEL_ID, [pdu.as_slice(), pdu.as_slice()].concat());
+
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10);
+    assert_eq!(process_frame(&mut processor, &frame).len(), 2);
+    let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35);
+    // There is no RDP Security Header: include TPKT, X224, MCS and both PDUs.
+    assert_eq!(
+        bandwidth_result(&result),
+        (2, 0x000b, 25, u32::try_from(frame.len()).unwrap() + 6)
+    );
+}
+
+#[test]
+fn continuous_measurement_counts_svc_framing() {
+    let mut channels = StaticChannelSet::new();
+    channels.insert(DrdynvcClient::new());
+    channels.attach_channel_id(core::any::TypeId::of::<DrdynvcClient>(), 1005);
+    let mut processor = Processor::new(
+        channels,
+        USER_CHANNEL_ID,
+        IO_CHANNEL_ID,
+        Some(MESSAGE_CHANNEL_ID),
+        SHARE_ID,
+    );
+    let capabilities = encode_vec(&DrdynvcServerPdu::Capabilities(CapabilitiesRequestPdu::new(
+        CapsVersion::V1,
+        None,
+    )))
+    .unwrap();
+    let mut chunk = encode_vec(&ChannelPduHeader {
+        length: u32::try_from(capabilities.len()).unwrap(),
+        flags: ChannelControlFlags::FLAG_FIRST | ChannelControlFlags::FLAG_LAST,
+    })
+    .unwrap();
+    chunk.extend_from_slice(&capabilities);
+    let frame = encode_server_data(1005, chunk);
+
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10);
+    process_frame(&mut processor, &frame);
+    let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35);
+    assert_eq!(
+        bandwidth_result(&result),
+        (2, 0x000b, 25, u32::try_from(frame.len()).unwrap() + 6)
+    );
+}
+
+#[test]
+fn continuous_measurement_excludes_io_multitransport_security_header() {
+    let mut processor = make_processor();
+    let data = encode_vec(&MultitransportRequestPdu {
+        security_header: BasicSecurityHeader {
+            flags: BasicSecurityHeaderFlags::TRANSPORT_REQ,
+        },
+        request_id: 42,
+        requested_protocol: RequestedProtocol::UdpFecR,
+        security_cookie: [0xab; 16],
+    })
+    .unwrap();
+    let counted = u32::try_from(data.len() - BasicSecurityHeader::FIXED_PART_SIZE).unwrap();
+    let frame = encode_server_data(IO_CHANNEL_ID, data);
+
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10);
+    assert!(process_frame(&mut processor, &frame).is_empty());
+    let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), 35);
+    assert_eq!(bandwidth_result(&result), (2, 0x000b, 25, counted + 6));
+}
+
+#[test]
+fn connect_time_measurement_includes_payload_headers_once() {
+    let mut processor = make_processor();
+    timed_request(&mut processor, AutoDetectRequest::bw_start_connect_time(1), 100);
+    timed_request(&mut processor, AutoDetectRequest::bw_payload(2, vec![0xaa; 64]), 110);
+    timed_request(&mut processor, AutoDetectRequest::rtt_continuous(3), 112);
+    let result = timed_request(
+        &mut processor,
+        AutoDetectRequest::bw_stop_connect_time(4, vec![0xbb; 16]),
+        140,
+    );
+    assert_eq!(bandwidth_result(&result), (4, 0x0003, 40, 64 + 8 + 16 + 8));
+}
+
+#[test]
+fn repeated_start_resets_count_and_timer_and_stop_ends_window() {
+    let mut processor = make_processor();
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 100);
+    timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 110);
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(3), 200);
+    let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(4), 210);
+    assert_eq!(bandwidth_result(&result), (4, 0x000b, 10, 6));
+    let repeated = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(5), 220);
+    assert_eq!(bandwidth_result(&repeated), (5, 0x000b, 1, 0));
+}
+
+#[test]
+fn measurement_timing_saturates_and_never_reports_zero_divisor() {
+    for (start, stop, expected) in [(100, 100, 1), (100, 90, 1), (0, u64::MAX, u32::MAX)] {
+        let mut processor = make_processor();
+        timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), start);
+        let result = timed_request(&mut processor, AutoDetectRequest::bw_stop_continuous(2), stop);
+        assert_eq!(bandwidth_result(&result), (2, 0x000b, expected, 6));
+    }
+}
+
+#[test]
+fn lossy_requests_on_main_channel_are_not_answered() {
+    use ironrdp_pdu::rdp::autodetect::{BW_START_LOSSY_UDP, BW_STOP_LOSSY_UDP};
+    let mut processor = make_processor();
+    assert!(
+        timed_request(
+            &mut processor,
+            AutoDetectRequest::BandwidthMeasureStart {
+                sequence_number: 1,
+                request_type: BW_START_LOSSY_UDP,
+            },
+            10
+        )
+        .is_empty()
+    );
+    assert!(
+        timed_request(
+            &mut processor,
+            AutoDetectRequest::BandwidthMeasureStop {
+                sequence_number: 1,
+                request_type: BW_STOP_LOSSY_UDP,
+                payload: None,
+            },
+            20
+        )
+        .is_empty()
+    );
+}
+
+#[test]
+fn untimed_driver_does_not_report_accumulated_bytes_as_a_real_measurement() {
+    let mut processor = make_processor();
+    timed_request(&mut processor, AutoDetectRequest::bw_start_continuous(1), 10);
+    timed_request(&mut processor, AutoDetectRequest::rtt_continuous(2), 20);
+    let outputs = process_frame(
+        &mut processor,
+        &encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(3)),
+    );
+    assert_eq!(bandwidth_result(&outputs), (3, 0x000b, 1, 0));
+}
+
+#[test]
+fn untimed_connect_time_stop_does_not_report_stop_payload_bytes() {
+    let mut processor = make_processor();
+    timed_request(&mut processor, AutoDetectRequest::bw_start_connect_time(1), 10);
+    timed_request(&mut processor, AutoDetectRequest::bw_payload(2, vec![0xaa; 64]), 20);
+    let outputs = process_frame(
+        &mut processor,
+        &encode_server_autodetect(AutoDetectRequest::bw_stop_connect_time(3, vec![0xbb; 16])),
+    );
+    assert_eq!(bandwidth_result(&outputs), (3, 0x0003, 1, 0));
+}
+
+#[test]
+fn connect_time_stop_without_a_timed_start_reports_zero_bytes() {
+    for send_untimed_start in [false, true] {
+        for stop_time in [None, Some(ironrdp_core::MonotonicInstant::from_millis(20))] {
+            let mut processor = make_processor();
+            if send_untimed_start {
+                process_frame(
+                    &mut processor,
+                    &encode_server_autodetect(AutoDetectRequest::bw_start_connect_time(1)),
+                );
+            }
+            let frame = encode_server_autodetect(AutoDetectRequest::bw_stop_connect_time(2, vec![0xbb; 16]));
+            let outputs = processor.process_with_timestamp(&frame, &mut None, stop_time).unwrap();
+            assert_eq!(bandwidth_result(&outputs), (2, 0x0003, 1, 0));
+        }
+    }
+}
+
+fn make_active_stage() -> ActiveStage {
+    ActiveStageBuilder {
+        static_channels: StaticChannelSet::new(),
+        user_channel_id: USER_CHANNEL_ID,
+        io_channel_id: IO_CHANNEL_ID,
+        message_channel_id: Some(MESSAGE_CHANNEL_ID),
+        share_id: SHARE_ID,
+        compression_type: None,
+        enable_server_pointer: false,
+        pointer_software_rendering: false,
+    }
+    .build()
+}
+
+/// A fast-path frame carrying one Synchronize update with `data_len` bytes of data.
+fn fast_path_frame(data_len: usize) -> Vec<u8> {
+    let data = vec![0; data_len];
+    let update = encode_vec(&FastPathUpdatePdu {
+        fragmentation: Fragmentation::Single,
+        update_code: UpdateCode::Synchronize,
+        compression_flags: None,
+        compression_type: None,
+        data: &data,
+    })
+    .unwrap();
+    let mut frame = encode_vec(&FastPathHeader::new(EncryptionFlags::empty(), update.len())).unwrap();
+    frame.extend_from_slice(&update);
+    frame
+}
+
+fn process_stage_frame(
+    stage: &mut ActiveStage,
+    image: &mut DecodedImage,
+    action: Action,
+    frame: &[u8],
+    millis: u64,
+) -> Vec<ActiveStageOutput> {
+    stage
+        .process_with_timestamp(
+            image,
+            action,
+            frame,
+            Some(ironrdp_core::MonotonicInstant::from_millis(millis)),
+        )
+        .expect("process timed frame")
+}
+
+#[test]
+fn continuous_measurement_counts_fast_path_data_only_inside_the_window() {
+    let mut stage = make_active_stage();
+    let mut image = DecodedImage::new(PixelFormat::RgbA32, 64, 64);
+    let frame = fast_path_frame(100);
+
+    process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 5);
+    let start = encode_server_autodetect(AutoDetectRequest::bw_start_continuous(1));
+    process_stage_frame(&mut stage, &mut image, Action::X224, &start, 10);
+    process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 20);
+    let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(2));
+    let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 40);
+    process_stage_frame(&mut stage, &mut image, Action::FastPath, &frame, 50);
+
+    let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else {
+        panic!("expected exactly one bandwidth response, got {outputs:?}");
+    };
+    // The whole fast-path frame counts because it has no RDP Security Header.
+    // The Stop adds only its six-byte auto-detect header, after its Security Header.
+    assert_eq!(
+        bandwidth_result_frame(response),
+        (2, 0x000b, 30, u32::try_from(frame.len()).unwrap() + 6)
+    );
+}
+
+fn at(millis: u64) -> ironrdp_core::MonotonicInstant {
+    ironrdp_core::MonotonicInstant::from_millis(millis)
+}
+
+#[test]
+fn tunnel_auto_detect_answers_an_rtt_request() {
+    let mut stage = make_active_stage();
+    let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::rtt_continuous(7)], 10, at(1_000));
+    assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]);
+}
+
+/// Count received payload before processing control messages, as on the main connection:
+/// the Start resets its carrying PDU's count, while the Stop includes its carrying PDU.
+#[test]
+fn tunnel_auto_detect_counts_the_data_between_start_and_stop() {
+    let mut stage = make_active_stage();
+
+    assert!(
+        stage
+            .process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 7, at(1_000))
+            .is_empty()
+    );
+    assert!(stage.process_tunnel_auto_detect(Vec::new(), 100, at(1_010)).is_empty());
+    let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(3)], 50, at(1_040));
+
+    let [
+        AutoDetectResponse::BandwidthMeasureResults {
+            sequence_number,
+            time_delta_ms,
+            byte_count,
+            ..
+        },
+    ] = responses.as_slice()
+    else {
+        panic!("expected one bandwidth result, got {responses:?}");
+    };
+    assert_eq!((*sequence_number, *time_delta_ms, *byte_count), (3, 40, 100 + 50));
+}
+
+/// The tunnel and the main connection each keep their own measurement, so data received on one
+/// is not counted in the other's, even while both windows are open.
+#[test]
+fn tunnel_and_main_connection_measure_separately() {
+    let mut stage = make_active_stage();
+    let mut image = DecodedImage::new(PixelFormat::RgbA32, 64, 64);
+
+    let start = encode_server_autodetect(AutoDetectRequest::bw_start_continuous(1));
+    process_stage_frame(&mut stage, &mut image, Action::X224, &start, 1_000);
+    stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, at(1_000));
+
+    process_stage_frame(&mut stage, &mut image, Action::FastPath, &fast_path_frame(100), 1_010);
+    stage.process_tunnel_auto_detect(Vec::new(), 500, at(1_010));
+
+    let tunnel = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, at(1_020));
+    assert!(
+        matches!(
+            tunnel.as_slice(),
+            [AutoDetectResponse::BandwidthMeasureResults { byte_count: 500, .. }]
+        ),
+        "{tunnel:?}"
+    );
+
+    let stop = encode_server_autodetect(AutoDetectRequest::bw_stop_continuous(2));
+    let outputs = process_stage_frame(&mut stage, &mut image, Action::X224, &stop, 1_020);
+    let [ActiveStageOutput::ResponseFrame(response)] = outputs.as_slice() else {
+        panic!("expected exactly one bandwidth response, got {outputs:?}");
+    };
+    // The complete fast-path frame and the Stop's own six bytes, but none of the tunnel's 500.
+    let tcp_bytes = u32::try_from(fast_path_frame(100).len()).unwrap();
+    assert_eq!(bandwidth_result_frame(response), (2, 0x000b, 20, tcp_bytes + 6));
+}
diff --git a/crates/ironrdp-testsuite-extra/Cargo.toml b/crates/ironrdp-testsuite-extra/Cargo.toml
index 3f5a348296..bc4719bae2 100644
--- a/crates/ironrdp-testsuite-extra/Cargo.toml
+++ b/crates/ironrdp-testsuite-extra/Cargo.toml
@@ -30,7 +30,7 @@ ironrdp-async.path = "../ironrdp-async"
 ironrdp-agent.path = "../ironrdp-agent"
 ironrdp-daemon.path = "../ironrdp-daemon"
 ironrdp-bulk.path = "../ironrdp-bulk"
-ironrdp-client = { path = "../ironrdp-client", features = ["sound", "udp"] }
+ironrdp-client = { path = "../ironrdp-client", features = ["sound", "udp", "__test"] }
 ironrdp-core.path = "../ironrdp-core"
 ironrdp-dvc.path = "../ironrdp-dvc"
 ironrdp-dvc-pipe-proxy.path = "../ironrdp-dvc-pipe-proxy"
diff --git a/crates/ironrdp-testsuite-extra/tests/client/mod.rs b/crates/ironrdp-testsuite-extra/tests/client/mod.rs
index d4172e224b..b147cf0f12 100644
--- a/crates/ironrdp-testsuite-extra/tests/client/mod.rs
+++ b/crates/ironrdp-testsuite-extra/tests/client/mod.rs
@@ -2,3 +2,4 @@ mod config;
 mod input;
 mod output_channel;
 mod rail;
+mod tunnel_auto_detect;
diff --git a/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs
new file mode 100644
index 0000000000..4dc6345d8d
--- /dev/null
+++ b/crates/ironrdp-testsuite-extra/tests/client/tunnel_auto_detect.rs
@@ -0,0 +1,150 @@
+//! Auto-detect requests and responses travel on the UDP tunnel as tunnel sub-headers, whose
+//! own two bytes are the auto-detect structure's headerLength and headerTypeId ([MS-RDPEMT]
+//! 2.2.1.1.1).
+//!
+//! [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad
+
+use ironrdp_client::udp::{tunnel_auto_detect_requests, tunnel_auto_detect_sub_header};
+use ironrdp_core::encode_vec;
+use ironrdp_pdu::rdp::autodetect::{AutoDetectRequest, AutoDetectResponse};
+use ironrdp_rdpemt::{SubHeaderType, TunnelData, TunnelSubHeader};
+
+/// Windows frames an RTT Measure Request on the tunnel as `06 00 <seq> 01 00`. Request types the
+/// decoder does not model, and sub-headers that are not requests, are skipped.
+#[test]
+fn tunnel_sub_headers_decode_as_auto_detect_requests() {
+    let rtt: TunnelSubHeader = ironrdp_core::decode(&[0x06, 0x00, 0x07, 0x00, 0x01, 0x00]).expect("sub-header");
+    let unmodeled = TunnelSubHeader {
+        sub_header_type: SubHeaderType::AutoDetectRequest,
+        data: vec![0x01, 0x00, 0xFF, 0x7F],
+    };
+    let response = TunnelSubHeader {
+        sub_header_type: SubHeaderType::AutoDetectResponse,
+        data: vec![0x07, 0x00, 0x00, 0x00],
+    };
+
+    let requests = tunnel_auto_detect_requests(&[rtt, unmodeled, response]);
+    assert!(matches!(
+        requests.as_slice(),
+        [AutoDetectRequest::RttRequest { sequence_number: 7, .. }]
+    ));
+}
+
+/// The RTT response goes back as `06 01 <seq> 00 00`, alone in a Tunnel Data PDU with no data.
+#[test]
+fn auto_detect_responses_encode_as_tunnel_sub_headers() {
+    let response = AutoDetectResponse::RttResponse { sequence_number: 7 };
+    let sub_header = tunnel_auto_detect_sub_header(&response).expect("sub-header");
+    let pdu = encode_vec(&TunnelData {
+        sub_headers: vec![sub_header],
+        higher_layer_data: Vec::new(),
+    })
+    .expect("encode");
+    // TunnelData: action 2, payloadLength 0, headerLength 4 + 6, then the sub-header.
+    assert_eq!(pdu, [0x02, 0x00, 0x00, 0x0A, 0x06, 0x01, 0x07, 0x00, 0x00, 0x00]);
+}
+
+#[test]
+fn bandwidth_sub_headers_preserve_the_wire_layout() {
+    let start = TunnelSubHeader {
+        sub_header_type: SubHeaderType::AutoDetectRequest,
+        data: vec![9, 0, 0x14, 0],
+    };
+    assert_eq!(
+        tunnel_auto_detect_requests(&[start]),
+        [AutoDetectRequest::bw_start_continuous(9)]
+    );
+    let response = AutoDetectResponse::BandwidthMeasureResults {
+        sequence_number: 9,
+        response_type: 0x000b,
+        time_delta_ms: 40,
+        byte_count: 150,
+    };
+    let sub_header = tunnel_auto_detect_sub_header(&response).expect("bandwidth sub-header");
+    assert_eq!(
+        encode_vec(&sub_header).unwrap(),
+        [14, 1, 9, 0, 11, 0, 40, 0, 0, 0, 150, 0, 0, 0]
+    );
+}
+
+#[test]
+fn tunnel_failure_before_migration_withdraws_soft_sync_without_ending_the_session() {
+    let mut stage = make_stage(false);
+    let mut transport = None;
+    ironrdp_client::udp::disable_failed_tunnel(
+        &mut stage,
+        &mut transport,
+        ironrdp::session::general_err!("auto-detect reply send failed"),
+    )
+    .expect("unused tunnel must fall back to TCP");
+    assert!(transport.is_none());
+    assert!(!stage.reliable_udp_dvc_tunnel_in_use());
+
+    // The failed tunnel must no longer be offered to a later Soft-Sync request.
+    use ironrdp_svc::SvcProcessor as _;
+    let drdynvc = stage.get_svc_processor_mut::<ironrdp_dvc::DrdynvcClient>().unwrap();
+    assert!(drdynvc.process(&soft_sync_request()).is_err());
+}
+
+#[test]
+fn tunnel_failure_after_migration_requires_reconnection() {
+    let mut stage = make_stage(true);
+    let mut transport = None;
+    let error = ironrdp_client::udp::disable_failed_tunnel(
+        &mut stage,
+        &mut transport,
+        ironrdp::session::general_err!("auto-detect reply send failed"),
+    )
+    .expect_err("migrated channels cannot silently resume on TCP");
+    assert!(error.to_string().contains("auto-detect reply send failed"));
+    assert!(stage.reliable_udp_dvc_tunnel_in_use());
+}
+
+fn make_stage(migrated: bool) -> ironrdp::session::ActiveStage {
+    use ironrdp_dvc::pdu::SoftSyncTunnelType;
+    use ironrdp_svc::SvcProcessor as _;
+
+    let mut drdynvc = ironrdp_dvc::DrdynvcClient::new();
+    drdynvc.attach_established_dynamic_channel(7, TestChannel).unwrap();
+    drdynvc.enable_soft_sync_tunnel(SoftSyncTunnelType::RELIABLE_UDP);
+    if migrated {
+        drdynvc.process(&soft_sync_request()).unwrap();
+    }
+    let mut static_channels = ironrdp_svc::StaticChannelSet::new();
+    static_channels.insert(drdynvc);
+    ironrdp::session::ActiveStageBuilder {
+        static_channels,
+        user_channel_id: 1001,
+        io_channel_id: 1003,
+        message_channel_id: Some(1005),
+        share_id: 1,
+        compression_type: None,
+        enable_server_pointer: false,
+        pointer_software_rendering: false,
+    }
+    .build()
+}
+
+fn soft_sync_request() -> Vec<u8> {
+    use ironrdp_dvc::pdu::{DrdynvcServerPdu, SoftSyncChannelList, SoftSyncRequestPdu, SoftSyncTunnelType};
+    encode_vec(&DrdynvcServerPdu::SoftSyncRequest(SoftSyncRequestPdu::new(vec![
+        SoftSyncChannelList::new(SoftSyncTunnelType::RELIABLE_UDP, vec![7]),
+    ])))
+    .unwrap()
+}
+
+struct TestChannel;
+ironrdp_core::impl_as_any!(TestChannel);
+
+impl ironrdp_dvc::DvcProcessor for TestChannel {
+    fn channel_name(&self) -> &str {
+        "test-channel"
+    }
+    fn start(&mut self, _channel_id: u32) -> ironrdp_pdu::PduResult<Vec<ironrdp_dvc::DvcMessage>> {
+        Ok(Vec::new())
+    }
+    fn process(&mut self, _channel_id: u32, _payload: &[u8]) -> ironrdp_pdu::PduResult<Vec<ironrdp_dvc::DvcMessage>> {
+        Ok(Vec::new())
+    }
+}
+impl ironrdp_dvc::DvcClientProcessor for TestChannel {}
diff --git a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs
index b207ba923e..a0069ef8c0 100644
--- a/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs
+++ b/crates/ironrdp-testsuite-extra/tests/rdpeudp_tokio.rs
@@ -11,10 +11,11 @@ use core::sync::atomic::{AtomicUsize, Ordering};
 use core::time::Duration;
 use std::sync::Arc;
 
-use ironrdp_rdpemt::TunnelConfig;
+use ironrdp_rdpemt::{SubHeaderType, TunnelConfig, TunnelSubHeader};
 use ironrdp_rdpeudp::ConnectionConfig;
 use ironrdp_rdpeudp_tokio::{
-    MultitransportBootstrap, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig, accept_udp, connect_udp,
+    MultitransportBootstrap, TunnelMessage, UdpAcceptConfig, UdpTlsConfig, UdpTransport, UdpTransportConfig,
+    accept_udp, connect_udp,
 };
 use ironrdp_tls::{CertificateValidation, CertificateValidationCallback};
 use tokio::net::UdpSocket;
@@ -222,6 +223,44 @@ async fn full_stack_bidirectional_data() {
     server.shutdown().await.expect("server shutdown");
 }
 
+/// Sub-headers cross the full stack beside their data, and a message may
+/// carry sub-headers alone, as an auto-detect Start or Stop does.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn full_stack_sub_headers() {
+    let (mut client, server) = establish_loopback_pair_on("127.0.0.1:0").await;
+
+    let start = TunnelMessage {
+        sub_headers: vec![TunnelSubHeader {
+            sub_header_type: SubHeaderType::AutoDetectRequest,
+            // A Bandwidth Measure Start: sequenceNumber 7, requestType 0x0014. The
+            // sub-header's own two bytes are its headerLength and headerTypeId.
+            data: vec![0x07, 0x00, 0x14, 0x00],
+        }],
+        data: Vec::new(),
+    };
+    let graphics = TunnelMessage {
+        sub_headers: vec![TunnelSubHeader {
+            sub_header_type: SubHeaderType::AutoDetectRequest,
+            data: vec![0x08],
+        }],
+        data: vec![0x01, 0x02, 0x03],
+    };
+    server
+        .send_message(start.clone())
+        .await
+        .expect("send sub-headers alone");
+    server
+        .send_message(graphics.clone())
+        .await
+        .expect("send sub-headers with data");
+
+    assert_eq!(client.recv_message().await.expect("recv"), start);
+    assert_eq!(client.recv_message().await.expect("recv"), graphics);
+
+    client.shutdown().await.expect("client shutdown");
+    server.shutdown().await.expect("server shutdown");
+}
+
 /// A payload over the wire `PayloadLength` field's 65535-byte capacity must
 /// be rejected synchronously by `send()`, and must not take the write pump
 /// down with it: a normal-sized send right after must still succeed.