@@ -50,6 +50,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 +83,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
 
@@ -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")]
@@ -2855,6 +2856,56 @@ fn build_ordinary_rdcleanpath_request(
 
 // ── Active session ────────────────────────────────────────────────────────────
 
+/// Decodes the auto-detect requests among the sub-headers of a Tunnel Data PDU.
+///
+/// Such a sub-header is the request structure itself: its SubHeaderLength and SubHeaderType
+/// are the request's headerLength and headerTypeId ([MS-RDPEMT] 2.2.1.1.1), so the request is
+/// decoded from the whole sub-header. Windows also sends request types the decoder does not
+/// model; those are skipped.
+///
+/// [MS-RDPEMT]: https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-rdpemt/4f538fd7-3aca-4e7d-a213-13eb5f95c1ad
+#[cfg(feature = "udp")]
+fn tunnel_auto_detect_requests(sub_headers: &[ironrdp_rdpemt::TunnelSubHeader]) -> Vec<AutoDetectRequest> {
+    sub_headers
+        .iter()
+        .filter(|sub_header| sub_header.sub_header_type == ironrdp_rdpemt::SubHeaderType::AutoDetectRequest)
+        .filter_map(|sub_header| {
+            let decoded = ironrdp_core::encode_vec(sub_header)
+                .map_err(|error| error.to_string())
+                .and_then(|structure| {
+                    ironrdp_core::decode::<AutoDetectRequest>(&structure).map_err(|error| error.to_string())
+                });
+            match decoded {
+                Ok(request) => Some(request),
+                Err(error) => {
+                    debug!(%error, data = ?sub_header.data, "Ignoring an undecodable auto-detect request on the tunnel");
+                    None
+                }
+            }
+        })
+        .collect()
+}
+
+/// Encodes an auto-detect response as the tunnel sub-header it is on the wire (see
+/// [`tunnel_auto_detect_requests`]).
+#[cfg(feature = "udp")]
+fn tunnel_auto_detect_sub_header(
+    response: &ironrdp_pdu::rdp::autodetect::AutoDetectResponse,
+) -> Option<ironrdp_rdpemt::TunnelSubHeader> {
+    let encoded = ironrdp_core::encode_vec(response)
+        .map_err(|error| error.to_string())
+        .and_then(|structure| {
+            ironrdp_core::decode::<ironrdp_rdpemt::TunnelSubHeader>(&structure).map_err(|error| error.to_string())
+        });
+    match encoded {
+        Ok(sub_header) => Some(sub_header),
+        Err(error) => {
+            debug!(%error, ?response, "Could not encode an auto-detect response for the tunnel");
+            None
+        }
+    }
+}
+
 enum RdpControlFlow {
     ReconnectWithNewSize {
         width: u16,
@@ -3085,6 +3136,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()?)
@@ -3228,22 +3282,25 @@ 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(
@@ -3258,12 +3315,45 @@ async fn active_session(
                         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(),
+                                Some(received_at),
+                            );
+                            let sub_headers: Vec<_> = responses.iter().filter_map(tunnel_auto_detect_sub_header).collect();
+                            if let (false, Some(transport)) = (sub_headers.is_empty(), 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 {
+                                    return Ok(RdpControlFlow::TransportFailure(ironrdp_session::custom_err!(
+                                        "answer reliable UDP tunnel auto-detect",
+                                        error
+                                    )));
+                                }
+                            }
+                        }
+                        #[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() {
                             let batch =
                                 active_stage.process_dvc_tunnel(SoftSyncTunnelType::RELIABLE_UDP, &payload)?;
                             ActiveSessionIteration::dvc(batch)
@@ -4386,12 +4476,51 @@ fn process_clipboard_message(
 mod tests {
     use super::*;
 
+    /// Windows frames an RTT Measure Request on the tunnel as `06 00 <seq> 01 00`: the
+    /// sub-header's own two bytes are the request's header.
+    #[cfg(feature = "udp")]
+    #[test]
+    fn tunnel_sub_headers_decode_as_auto_detect_requests() {
+        use ironrdp_rdpemt::{SubHeaderType, TunnelSubHeader};
+
+        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.
+    #[cfg(feature = "udp")]
+    #[test]
+    fn auto_detect_responses_encode_as_tunnel_sub_headers() {
+        let response = ironrdp_pdu::rdp::autodetect::AutoDetectResponse::RttResponse { sequence_number: 7 };
+        let sub_header = tunnel_auto_detect_sub_header(&response).expect("sub-header");
+        let pdu = encode_vec(&ironrdp_rdpemt::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]);
+    }
+
     #[cfg(feature = "rdpdr")]
     use core::any::TypeId;
     #[cfg(feature = "rdpdr")]
     use core::sync::atomic::AtomicUsize;
 
-    #[cfg(feature = "rdpdr")]
+    #[cfg(any(feature = "rdpdr", feature = "udp"))]
     use ironrdp_core::encode_vec;
     use ironrdp_pdu::input::fast_path::KeyboardFlags;
     #[cfg(feature = "rdpdr")]
 
@@ -11,7 +11,7 @@ use ironrdp_pdu::fast_path::FastPathHeader;
 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;
@@ -28,6 +28,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};
@@ -50,6 +51,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`].
@@ -108,6 +111,7 @@ impl ActiveStageBuilder {
             window_support_level: None,
             graphics_output_needs_full_refresh: false,
             damage_regions: Vec::new(),
+            tunnel_auto_detect: AutoDetectResponder::default(),
         }
     }
 }
@@ -575,6 +579,33 @@ impl ActiveStage {
             .is_some_and(|drdynvc| drdynvc.has_channels_on_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). That data follows the sub-headers on the wire, so the data of the
+    /// PDU that carries a Start is counted and the data of the one that carries a Stop is not.
+    /// `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: Option<MonotonicInstant>,
+    ) -> Vec<AutoDetectResponse> {
+        let responses = requests
+            .into_iter()
+            .filter_map(|request| self.tunnel_auto_detect.respond(request, received_at))
+            .collect();
+        self.tunnel_auto_detect.record_bytes(data_len);
+        responses
+    }
+
     /// 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
@@ -1079,6 +1110,76 @@ mod tests {
     use ironrdp_pdu::pointer::{ColorPointerAttribute, Point16, PointerAttribute, PointerUpdateData};
     use ironrdp_rdpei::pdu::{PenEventPdu, RdpInputProtocolVersion, RdpeiPdu, ScReadyPdu, TouchEventPdu};
 
+    fn tunnel_test_stage() -> ActiveStage {
+        ActiveStageBuilder {
+            static_channels: StaticChannelSet::new(),
+            user_channel_id: 1001,
+            io_channel_id: 1003,
+            message_channel_id: None,
+            share_id: 1,
+            compression_type: None,
+            enable_server_pointer: true,
+            pointer_software_rendering: false,
+        }
+        .build()
+    }
+
+    #[test]
+    fn tunnel_auto_detect_answers_an_rtt_request() {
+        let mut stage = tunnel_test_stage();
+        let responses = stage.process_tunnel_auto_detect(
+            vec![AutoDetectRequest::rtt_continuous(7)],
+            10,
+            Some(MonotonicInstant::from_millis(1_000)),
+        );
+        assert_eq!(responses, [AutoDetectResponse::RttResponse { sequence_number: 7 }]);
+    }
+
+    /// The data of the PDU that carries the Start follows it on the wire and is counted; the data
+    /// of the one that carries the Stop is not.
+    #[test]
+    fn tunnel_auto_detect_counts_the_data_between_start_and_stop() {
+        let mut stage = tunnel_test_stage();
+        let at = |millis| Some(MonotonicInstant::from_millis(millis));
+
+        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, 7 + 100));
+    }
+
+    /// The tunnel's measurement is separate from the message channel's, so data that arrives over
+    /// TCP is not counted in it.
+    #[test]
+    fn tunnel_auto_detect_keeps_its_own_window() {
+        let mut stage = tunnel_test_stage();
+        let at = |millis| Some(MonotonicInstant::from_millis(millis));
+
+        stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_start_continuous(1)], 0, at(1_000));
+        stage.x224_processor.record_bandwidth_bytes(500);
+        let responses = stage.process_tunnel_auto_detect(vec![AutoDetectRequest::bw_stop_continuous(2)], 0, at(1_020));
+        assert!(matches!(
+            responses.as_slice(),
+            [AutoDetectResponse::BandwidthMeasureResults { byte_count: 0, .. }]
+        ));
+    }
+
     #[test]
     fn full_redraw_prefers_suppress_output_toggle_when_supported() {
         let stage = ActiveStageBuilder {