Expand VLESS interoperability coverage and fix XHTTP shutdown (#189)

Expand legacy and encrypted VLESS coverage across peer transports, Vision transitions, session lifecycle, replay isolation, and chained connections, with Xray updated to 26.10.17. Preserve complete XHTTP HTTP/2 responses when peers gracefully stop an unfinished upload while continuing to reject truncated responses and incomplete packet uploads. Isolate the pinned sing-box server TLS 1.2 shutdown limitation in a test-only workaround, retaining an ignored reproducer and a reference to SagerNet/sing-vmess#16 for removal when the peer pin includes the fix.

.github/workflows/reality-interop.yml

@@ -16,8 +16,8 @@ jobs:
         include:
           - name: v26.10.10
             revision: 701af60772cda123492f86e383e1fdba066614a9
-          - name: main-2026-10-14
-            revision: 0888beb38a9ba45440557175e71ce21cf8ba3b13
+          - name: v26.10.17
+            revision: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5
     name: Xray ${{ matrix.name }}
     env:
       CARGO_BUILD_JOBS: 1

.github/workflows/test.yml

@@ -13,9 +13,9 @@ env:
   SING_BOX_SHA256: 12cb2816b52febb356f6a885b740cc8758c3f30b8ae0ca8edba80f0d2d35343f
   MIHOMO_VERSION: 1.19.31
   MIHOMO_SHA256: 04cf9f09671704f839ddbee2e93069dc831a4123a75281e725d1d96ab9ac1afc
-  XRAY_VERSION: 26.3.27
-  XRAY_SHA256: 23cd9af937744d97776ee35ecad4972cf4b2109d1e0fe6be9930467608f7c8ae
-  XRAY_COMMIT: 701af60772cda123492f86e383e1fdba066614a9
+  XRAY_VERSION: 26.10.17
+  XRAY_SHA256: ecdd54eccd24c4c21a8f00968e7dbf7f5996da92383d2321170b560dc4a95158
+  XRAY_COMMIT: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5
   MIHOMO_ALPHA_COMMIT: 94b13aed49b8dc41973e4bb3f721c5fa2c5c5c1f
   NAIVE_CADDY_VERSION: 2.11.7-naive
   NAIVE_CADDY_SHA256: ac97057c925bb098eb575e7b0cb8bd547d4c5f845b09a3dedfbe34444aa7ae88
@@ -45,7 +45,7 @@ jobs:
     env:
       GOMAXPROCS: 1
       WG_SING_BOX_COMMIT: 7054cac5124465ffdb145870d7c01885566711b1
-      WG_XRAY_COMMIT: 0888beb38a9ba45440557175e71ce21cf8ba3b13
+      WG_XRAY_COMMIT: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5
     steps:
     - uses: actions/checkout@v4
     - uses: dtolnay/rust-toolchain@stable

.github/workflows/vless-encryption-interop.yml

@@ -25,7 +25,7 @@ jobs:
       - uses: actions/checkout@v4
         with:
           repository: XTLS/Xray-core
-          ref: 0f3d884786f9ec73d3be8899cb6b9e1e6dd3c41e
+          ref: 92ccc939aaf4f8965f369cf0b142e005aeaa11d5
           path: peers/xray
       - uses: actions/checkout@v4
         with:
@@ -69,6 +69,7 @@ jobs:
         env:
           XRAY_BINARY: ${{ runner.temp }}/xray
           MIHOMO_BINARY: ${{ runner.temp }}/mihomo
+          SHOES_TEST_SING_BOX_BIN: ${{ runner.temp }}/sing-box
         run: cargo test --locked --features optional-peer-tests --test vless_encryption_interop -- --test-threads=1
       - name: Cross-language cryptography vectors
         run: |

README.md

@@ -557,14 +557,14 @@ covering all three modes and HTTP versions. `type: splithttp` is an alias for
   transport setup (or the shorter configured setup timeout). H1 keep-alive idle
   periods and subsequent partial headers have no deadline. For exposed listeners,
   set `global_limits.max_connections` and `global_limits.max_connections_per_ip`.
-- Xray 26.10.10's H1 packet-up uploader can lose pipelined requests when Go GC
-  discards pooled connections with unread responses. Sustained uploads to slow
-  destinations can therefore fail despite bounded server backpressure. Prefer
-  H2/H3 or a streaming mode for these clients; disabling GC is not a production fix.
+- Xray's H1 packet-up uploader (observed on 26.10.10) can lose pipelined requests
+  when Go GC discards pooled connections with unread responses. Sustained uploads
+  to slow destinations can therefore fail despite bounded server backpressure.
+  Prefer H2/H3 or a streaming mode for these clients; disabling GC is not a production fix.
 - Advanced placement/padding obfuscation, custom session IDs, separate downloads,
-  full XMUX policy, and Vision over XHTTP are not supported. Unsupported options
-  are rejected rather than silently ignored. Public CDN compatibility is not implied.
-- REALITY XHTTP interoperates with Xray 26.10.10 in both directions over H2,
+  full XMUX policy, and legacy Vision over XHTTP are not supported. Unsupported
+  options are rejected rather than silently ignored. Public CDN compatibility is not implied.
+- REALITY XHTTP interoperates with Xray 26.10.17 in both directions over H2,
   including all three modes and `auto`. REALITY prefers X25519MLKEM768 with
   X25519 fallback according to the cover target. Optional `mldsa65_seed` and
   `mldsa65_verify` apply to the outer REALITY configuration, independently of

src/vless/encryption/handshake.rs

@@ -155,6 +155,7 @@ impl ClientEncryption {
                     used,
                 },
             )?;
+            log::debug!("VLESS encryption client: resumed handshake");
             return Ok(stream);
         }
         let mlkem = kem::DecapsulationKey::generate(&kem::ML_KEM_768)
@@ -276,6 +277,7 @@ impl ClientEncryption {
                 .map_err(|_| {
                     io::Error::new(io::ErrorKind::TimedOut, "VLESS fresh handshake timed out")
                 })??;
+        log::debug!("VLESS encryption client: fresh handshake");
         Ok(EncryptionStream::new(io, true, keys, initial))
     }
 
@@ -319,6 +321,14 @@ impl ServerEncryption {
         if let Some(random) = random_prefix {
             stream.set_server_random(random);
         }
+        log::debug!(
+            "VLESS encryption server: {} handshake",
+            if random_prefix.is_some() {
+                "resumed"
+            } else {
+                "fresh"
+            }
+        );
         Ok(stream)
     }
 
@@ -345,6 +355,7 @@ impl ServerEncryption {
         };
         let length = u16::from_be_bytes(plain.try_into().unwrap()) as usize;
         let cipher = nfs_aead.cipher;
+        log::debug!("VLESS encryption server cipher: {cipher:?}");
         let random_mode = self.mode == Mode::Random;
         if length == 32 {
             let mut encrypted_ticket = [0; 32];

src/vless/encryption/mod.rs

@@ -8,7 +8,7 @@ mod session;
 mod stream;
 
 #[cfg(test)]
-mod tests;
+pub(crate) mod tests;
 
 use std::{fmt, io, sync::Arc};
 

src/vless/encryption/stream.rs

@@ -131,7 +131,7 @@ pub(super) struct StreamKeys {
 
 impl<S: AsyncStream> EncryptionStream<S> {
     #[cfg(test)]
-    pub(super) fn is_resumed(&self) -> bool {
+    pub(crate) fn is_resumed(&self) -> bool {
         self.resume.is_some()
     }
     pub(super) fn new(inner: S, client: bool, keys: StreamKeys, initial: InitialRead) -> Self {

src/vless/encryption/stream/tests.rs

@@ -7,6 +7,54 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 mod liveness;
 mod resumption;
 
+#[tokio::test]
+async fn maximum_native_record_keeps_direct_command_in_second_vision_read_and_raw_tail() {
+    use crate::vless::vision_stream::VisionStream;
+    for masked in [false, true] {
+        for cipher in [Cipher::Aes256Gcm, Cipher::ChaCha20Poly1305] {
+            let first = vec![41; 9000];
+            let second = vec![42; 7596];
+            let mut plain = vec![0, 0];
+            plain.extend_from_slice(&[7; 16]);
+            plain.push(0);
+            plain.extend_from_slice(&(first.len() as u16).to_be_bytes());
+            plain.extend_from_slice(&[0, 0]);
+            plain.extend_from_slice(&first);
+            plain.push(2);
+            plain.extend_from_slice(&(second.len() as u16).to_be_bytes());
+            plain.extend_from_slice(&[0, 0]);
+            plain.extend_from_slice(&second);
+            assert_eq!(plain.len(), 16624);
+            let mut peer = Aead::new(b"context", &[7; 96], cipher).unwrap();
+            let mut wire = peer.seal_record(&plain, &[7; 96]).unwrap();
+            wire.extend_from_slice(b"\x17\x03\x03\x00\x04tail");
+            if masked {
+                HeaderMask::new(&[7; 96], &[8; 16], 0)
+                    .unwrap()
+                    .apply_write(&mut wire)
+                    .unwrap();
+            }
+            let mut stream = stream(
+                Scripted {
+                    input: wire.into(),
+                    eof: true,
+                    ..Default::default()
+                },
+                true,
+                masked,
+            );
+            stream.read_aead = Some(Aead::new(b"context", &[7; 96], cipher).unwrap());
+            let mut vision = VisionStream::new_client(stream, [7; 16]).unwrap();
+            let mut received = Vec::new();
+            vision.read_to_end(&mut received).await.unwrap();
+            assert_eq!(
+                received,
+                [first, second, b"\x17\x03\x03\x00\x04tail".to_vec()].concat()
+            );
+        }
+    }
+}
+
 #[derive(Default)]
 struct Scripted {
     input: VecDeque<u8>,

src/vless/encryption/tests.rs

@@ -8,6 +8,7 @@ use super::{
 };
 
 mod lifecycle;
+mod listeners;
 mod transports;
 
 pub(super) fn fixtures() -> Value {
@@ -164,7 +165,7 @@ pub(super) fn fixture_settings(mode: &str, resume: bool, keys: &[usize]) -> (Str
     )
 }
 
-pub(super) fn fixture_pair(
+pub(crate) fn fixture_pair(
     mode: &str,
     resume: bool,
     keys: &[usize],

src/vless/encryption/tests/listeners.rs

@@ -0,0 +1,253 @@
+use std::{io, sync::Arc, time::Duration};
+
+use serde_json::json;
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpListener, TcpStream},
+    time::timeout,
+};
+
+use super::{
+    fixture_settings,
+    transports::{Tasks, echo, request},
+};
+use crate::{
+    async_stream::AsyncStream, config::ServerConfig, resolver::NativeResolver,
+    vless::encryption::ClientEncryption,
+};
+
+const LIMIT: Duration = Duration::from_secs(10);
+const UUID: &str = "550e8400-e29b-41d4-a716-446655440000";
+
+async fn port() -> u16 {
+    TcpListener::bind("0.0.0.0:0")
+        .await
+        .unwrap()
+        .local_addr()
+        .unwrap()
+        .port()
+}
+
+async fn listen(config: serde_json::Value) -> Tasks {
+    let config: ServerConfig = serde_json::from_value(config).unwrap();
+    Tasks(
+        crate::tcp::tcp_server::start_servers(
+            crate::config::Config::Server(config),
+            Arc::new(NativeResolver::new()),
+        )
+        .await
+        .unwrap(),
+    )
+}
+
+async fn exchange(stream: &mut (impl AsyncStream + ?Sized), target: u16) -> io::Result<()> {
+    stream.write_all(&request(target, 1, false)).await?;
+    stream.write_all(b"scope").await?;
+    stream.flush().await?;
+    let mut reply = [0; 7];
+    stream.read_exact(&mut reply).await?;
+    assert_eq!(&reply, b"\0\0scope");
+    Ok(())
+}
+
+#[tokio::test]
+async fn simultaneous_resumed_replays_across_bind_addresses_reach_destination_once() {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use tokio::{sync::Barrier, task::JoinSet};
+
+    timeout(LIMIT, async {
+        let (encryption, decryption) = fixture_settings("random", true, &[32]);
+        let client = ClientEncryption::new(&encryption).unwrap();
+        let port = port().await;
+        let mut listeners = listen(json!({
+            "addresses":[format!("127.0.0.1:{port}"),format!("127.0.0.2:{port}")],
+            "protocol":{"type":"vless","user_id":UUID,"decryption":decryption}
+        }))
+        .await;
+        let target = TcpListener::bind("0.0.0.0:0").await.unwrap();
+        let target_port = target.local_addr().unwrap().port();
+        let accepted = Arc::new(AtomicUsize::new(0));
+        let count = accepted.clone();
+        let mut destination = Tasks(vec![tokio::spawn(async move {
+            let mut tasks = JoinSet::new();
+            loop {
+                tokio::select! {
+                    result = target.accept() => {
+                        let (mut stream, _) = result.unwrap();
+                        count.fetch_add(1, Ordering::SeqCst);
+                        tasks.spawn(async move {
+                            let (mut r, mut w) = stream.split();
+                            let _ = tokio::io::copy(&mut r, &mut w).await;
+                        });
+                    }
+                    _ = tasks.join_next(), if !tasks.is_empty() => {}
+                }
+            }
+        })]);
+        let mut warm = client
+            .handshake(TcpStream::connect(("127.0.0.1", port)).await.unwrap())
+            .await
+            .unwrap();
+        exchange(&mut warm, target_port).await.unwrap();
+        drop(warm);
+        assert_eq!(accepted.load(Ordering::SeqCst), 1);
+
+        // Capture an unspent resumed flight, then race its identical bytes on both listeners.
+        let (a, mut b) = tokio::io::duplex(8192);
+        let mut resumed = client.handshake(a).await.unwrap();
+        assert!(resumed.is_resumed());
+        resumed
+            .write_all(&request(target_port, 1, false))
+            .await
+            .unwrap();
+        resumed.write_all(b"scope").await.unwrap();
+        resumed.shutdown().await.unwrap();
+        drop(resumed);
+        let mut flight = Vec::new();
+        b.read_to_end(&mut flight).await.unwrap();
+        let barrier = Arc::new(Barrier::new(16));
+        let mut tasks = JoinSet::new();
+        for i in 0..16 {
+            let barrier = barrier.clone();
+            let flight = flight.clone();
+            tasks.spawn(async move {
+                let mut stream =
+                    TcpStream::connect((if i % 2 == 0 { "127.0.0.1" } else { "127.0.0.2" }, port))
+                        .await
+                        .unwrap();
+                barrier.wait().await;
+                stream.write_all(&flight).await.unwrap();
+                stream.shutdown().await.unwrap();
+                let mut response = Vec::new();
+                // A rejected flight may reset TCP with unread application bytes.
+                // Wait for EOF/reset, not just the server's initial random prefix.
+                if let Err(error) = stream.read_to_end(&mut response).await {
+                    assert_eq!(error.kind(), io::ErrorKind::ConnectionReset);
+                }
+            });
+        }
+        while let Some(result) = tasks.join_next().await {
+            result.unwrap();
+        }
+        // A final echo drains the accept queue before the duplicate-count assertion.
+        let mut sentinel = TcpStream::connect(("127.0.0.1", target_port))
+            .await
+            .unwrap();
+        sentinel.write_all(b"done").await.unwrap();
+        sentinel.read_exact(&mut [0; 4]).await.unwrap();
+        assert_eq!(accepted.load(Ordering::SeqCst), 3);
+        listeners.stop().await;
+        destination.stop().await;
+    })
+    .await
+    .unwrap();
+}
+
+#[derive(Debug)]
+struct Carrier;
+
+#[async_trait::async_trait]
+impl crate::tcp::tcp_handler::TcpClientHandler for Carrier {
+    async fn setup_client_tcp_stream(
+        &self,
+        stream: Box<dyn AsyncStream>,
+        _: crate::address::ResolvedLocation,
+    ) -> io::Result<crate::tcp::tcp_handler::TcpClientSetupResult> {
+        Ok(crate::tcp::tcp_handler::TcpClientSetupResult {
+            client_stream: stream,
+            early_data: None,
+        })
+    }
+}
+
+#[tokio::test]
+async fn native_tickets_are_shared_across_bind_ips_but_isolated_by_sni_and_websocket_path() {
+    use crate::{
+        address::{Address, NetLocation, ResolvedLocation},
+        tcp::tcp_handler::TcpClientHandler,
+        tls_client_handler::TlsClientHandler,
+        websocket::WebsocketTcpClientHandler,
+    };
+    timeout(LIMIT, async {
+        for tls in [false, true] {
+            let (encryption, decryption) = fixture_settings("native", true, &[32]);
+            let client = ClientEncryption::new(&encryption).unwrap();
+            let cert =
+                rcgen::generate_simple_self_signed(vec!["a.local".into(), "b.local".into()]).unwrap();
+            let protocol = json!({"type":"vless","user_id":UUID,"decryption":decryption});
+            let outer = if tls {
+                let target = json!({"cert":cert.cert.pem(),"key":cert.signing_key.serialize_pem(),"protocol":protocol});
+                json!({"type":"tls","tls_targets":{"a.local":target,"b.local":target}})
+            } else {
+                json!({"type":"websocket","targets":[
+                    {"matching_path":"/a","protocol":protocol},
+                    {"matching_path":"/b","protocol":protocol}
+                ]})
+            };
+            let port = port().await;
+            let mut listeners = listen(json!({
+                "addresses":[format!("127.0.0.1:{port}"),format!("127.0.0.2:{port}")],"protocol":outer
+            }))
+            .await;
+            let (target, mut destination) = echo().await;
+            let mut roots = rustls::RootCertStore::empty();
+            roots.add(cert.cert.der().clone()).unwrap();
+            let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
+            let config = Arc::new(
+                rustls::ClientConfig::builder_with_provider(provider)
+                    .with_safe_default_protocol_versions()
+                    .unwrap()
+                    .with_root_certificates(roots)
+                    .with_no_client_auth(),
+            );
+            for (scope, ip, resumed, rejected) in [
+                ("a", 1, false, false),
+                ("a", 2, true, false),
+                ("b", 1, true, true),
+                ("b", 1, false, false),
+                ("b", 2, true, false),
+                ("a", 2, true, true),
+                ("a", 1, false, false),
+            ] {
+                let wrapper: Box<dyn TcpClientHandler> = if tls {
+                    Box::new(TlsClientHandler::new(
+                        config.clone(),
+                        None,
+                        format!("{scope}.local").try_into().unwrap(),
+                        Box::new(Carrier),
+                    ))
+                } else {
+                    Box::new(WebsocketTcpClientHandler::new(
+                        Some(format!("/{scope}")),
+                        None,
+                        Default::default(),
+                        Box::new(Carrier),
+                    ))
+                };
+                let socket = TcpStream::connect((std::net::Ipv4Addr::new(127, 0, 0, ip), port))
+                    .await
+                    .unwrap();
+                let location = ResolvedLocation::new(NetLocation::new(
+                    Address::Ipv4(std::net::Ipv4Addr::LOCALHOST),
+                    target,
+                ));
+                let carrier = wrapper
+                    .setup_client_tcp_stream(Box::new(socket), location)
+                    .await
+                    .unwrap()
+                    .client_stream;
+                let mut stream = client.handshake(carrier).await.unwrap();
+                assert_eq!(stream.is_resumed(), resumed);
+                assert_eq!(
+                    exchange(&mut stream, target).await.is_err(),
+                    rejected,
+                    "tls={tls}, scope={scope}, ip={ip}"
+                );
+            }
+            listeners.stop().await;
+            destination.stop().await;
+        }
+    })
+    .await
+    .unwrap();
+}

src/vless/encryption/tests/transports.rs

@@ -17,7 +17,7 @@ use crate::{
 const UUID: &str = "550e8400-e29b-41d4-a716-446655440000";
 const LIMIT: Duration = Duration::from_secs(10);
 
-struct Tasks(Vec<JoinHandle<()>>);
+pub(super) struct Tasks(pub(super) Vec<JoinHandle<()>>);
 impl Drop for Tasks {
     fn drop(&mut self) {
         for task in &self.0 {
@@ -26,7 +26,7 @@ impl Drop for Tasks {
     }
 }
 impl Tasks {
-    async fn stop(&mut self) {
+    pub(super) async fn stop(&mut self) {
         for task in self.0.drain(..) {
             task.abort();
             let _ = task.await;
@@ -34,7 +34,7 @@ impl Tasks {
     }
 }
 
-async fn echo() -> (u16, Tasks) {
+pub(super) async fn echo() -> (u16, Tasks) {
     let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
     let port = listener.local_addr().unwrap().port();
     let task = tokio::spawn(async move {
@@ -55,7 +55,207 @@ async fn echo() -> (u16, Tasks) {
     (port, Tasks(vec![task]))
 }
 
-fn request(port: u16, command: u8, flow: bool) -> Vec<u8> {
+#[tokio::test]
+async fn native_vision_rejects_h2mux_before_opening_a_stream() {
+    use crate::{
+        client_proxy_selector::ClientProxySelector,
+        h2mux::{MUX_DESTINATION_HOST, MUX_DESTINATION_PORT},
+        tcp::tcp_handler::TcpServerHandler,
+        vless::{encryption::ServerEncryption, vless_server_handler::VlessTcpServerHandler},
+    };
+    use shoes_test_support::vless::{VlessDestination, encode_vless_udp_request, parse_uuid};
+
+    let (encryption, decryption) = fixture_settings("random", true, &[32]);
+    let handler = VlessTcpServerHandler::new(
+        UUID,
+        true,
+        Arc::new(ClientProxySelector::new(Vec::new())),
+        Arc::new(NativeResolver::new()),
+        None,
+        Some(Arc::new(
+            ServerEncryption::new(&decryption, Default::default()).unwrap(),
+        )),
+        true,
+    )
+    .unwrap();
+    let (a, b) = tokio::io::duplex(8192);
+    let (result, _client) = timeout(LIMIT, async {
+        tokio::join!(handler.setup_server_stream(Box::new(b)), async {
+            let mut client = ClientEncryption::new(&encryption)
+                .unwrap()
+                .handshake(a)
+                .await
+                .unwrap();
+            let mut request = encode_vless_udp_request(
+                parse_uuid(UUID).unwrap(),
+                &VlessDestination::Domain(MUX_DESTINATION_HOST.into(), MUX_DESTINATION_PORT),
+            )
+            .unwrap();
+            request[18] = 1;
+            request[17] = 18;
+            request.splice(18..18, b"\x0a\x10xtls-rprx-vision".iter().copied());
+            client.write_all(&request).await.unwrap();
+            client.flush().await.unwrap();
+            client
+        })
+    })
+    .await
+    .unwrap();
+    let error = result.err().expect("native Vision accepted H2MUX");
+    assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+    assert_eq!(
+        error.to_string(),
+        "Native VLESS Vision does not support H2MUX"
+    );
+}
+
+#[tokio::test]
+async fn legacy_and_native_address_forms_and_empty_udp_responses() {
+    use crate::async_stream::AsyncStream;
+    use shoes_test_support::{
+        test_servers::{start_tcp_stream_echo_server, start_udp_response_server},
+        vless::{
+            VlessDestination, VlessUdpClient, XUDP_OPTION_DATA, XUDP_STATUS_NEW, XudpClient,
+            XudpFrame, encode_vless_udp_request, parse_uuid,
+        },
+    };
+
+    let udp_response = |payload: &[u8], _| {
+        if payload == b"empty reply" {
+            vec![]
+        } else {
+            payload.to_vec()
+        }
+    };
+    let udp = start_udp_response_server("0.0.0.0", 0, udp_response)
+        .await
+        .unwrap();
+    let (tcp_port, mut target) = echo().await;
+    // Separate listeners serve either localhost address without requiring dual-stack sockets.
+    let ipv6_targets = match TcpListener::bind("[::1]:0").await {
+        Ok(_probe) => Some((
+            start_tcp_stream_echo_server("::1", tcp_port).await.unwrap(),
+            start_udp_response_server("::1", udp.local_addr().port(), udp_response)
+                .await
+                .unwrap(),
+        )),
+        Err(error) => {
+            eprintln!("IPv6 loopback unavailable; skipping IPv6 cases: {error}");
+            None
+        }
+    };
+    for (native, vision) in [(false, false), (true, false), (true, true)] {
+        let (encryption, decryption) = fixture_settings("random", true, &[32]);
+        let client = ClientEncryption::new(&encryption).unwrap();
+        let port = TcpListener::bind("0.0.0.0:0")
+            .await
+            .unwrap()
+            .local_addr()
+            .unwrap()
+            .port();
+        let config: ServerConfig = serde_json::from_value(json!({
+            "address":format!("0.0.0.0:{port}"),
+            "protocol":{"type":"vless","user_id":UUID,"udp_enabled":true,
+                "decryption":if native {decryption.as_str()} else {"none"},"vision":vision}
+        }))
+        .unwrap();
+        let mut listeners = Tasks(
+            crate::tcp::tcp_server::start_servers(
+                crate::config::Config::Server(config),
+                Arc::new(NativeResolver::new()),
+            )
+            .await
+            .unwrap(),
+        );
+        let connect = || async {
+            let socket = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
+            let stream: Box<dyn AsyncStream> = if native {
+                Box::new(client.handshake(socket).await.unwrap())
+            } else {
+                Box::new(socket)
+            };
+            stream
+        };
+        timeout(LIMIT, async {
+            let destinations = |port, ipv6: Option<SocketAddr>| {
+                let mut destinations = vec![
+                    VlessDestination::Ip(SocketAddr::from(([127, 0, 0, 1], port))),
+                    VlessDestination::Domain("localhost".into(), port),
+                ];
+                if let Some(address) = ipv6 {
+                    destinations.push(VlessDestination::Ip(address));
+                }
+                destinations
+            };
+            for _ in 0..2 {
+                if !vision {
+                    for destination in destinations(
+                        tcp_port,
+                        ipv6_targets.as_ref().map(|(tcp, _)| tcp.local_addr()),
+                    ) {
+                        let mut stream = connect().await;
+                        let mut request =
+                            encode_vless_udp_request(parse_uuid(UUID).unwrap(), &destination)
+                                .unwrap();
+                        request[18] = 1;
+                        stream.write_all(&request).await.unwrap();
+                        stream.write_all(b"address roundtrip").await.unwrap();
+                        stream.flush().await.unwrap();
+                        let mut reply = [0; 19];
+                        stream.read_exact(&mut reply).await.unwrap();
+                        assert_eq!(&reply[..2], &[0, 0]);
+                        assert_eq!(&reply[2..], b"address roundtrip");
+                    }
+                }
+                for destination in destinations(
+                    udp.local_addr().port(),
+                    ipv6_targets.as_ref().map(|(_, udp)| udp.local_addr()),
+                ) {
+                    let stream = connect().await;
+                    let mut udp = VlessUdpClient::connect(
+                        stream,
+                        parse_uuid(UUID).unwrap(),
+                        destination.clone(),
+                    )
+                    .await
+                    .unwrap();
+                    for (payload, expected) in [
+                        (b"empty reply".as_slice(), b"".as_slice()),
+                        (
+                            b"nonempty after empty".as_slice(),
+                            b"nonempty after empty".as_slice(),
+                        ),
+                    ] {
+                        udp.send_packet(payload).await.unwrap();
+                        let reply = udp.recv_packet(LIMIT).await.unwrap();
+                        assert_eq!(reply, expected);
+                    }
+                    let mut xudp = XudpClient::connect(connect().await, parse_uuid(UUID).unwrap())
+                        .await
+                        .unwrap();
+                    xudp.send_frame(&XudpFrame {
+                        session_id: 17,
+                        status: XUDP_STATUS_NEW,
+                        options: XUDP_OPTION_DATA,
+                        destination: Some(destination),
+                        payload: Some(b"xudp address".to_vec()),
+                    })
+                    .await
+                    .unwrap();
+                    let reply = xudp.recv_frame(LIMIT).await.unwrap();
+                    assert_eq!(reply.session_id, 17);
+                    assert_eq!(reply.payload.unwrap(), b"xudp address");
+                }
+            }
+        })
+        .await
+        .unwrap_or_else(|e| panic!("native={native}, vision={vision}: {e}"));
+        listeners.stop().await;
+    }
+    target.stop().await;
+}
+
+pub(super) fn request(port: u16, command: u8, flow: bool) -> Vec<u8> {
     let mut request = vec![0];
     request.extend(crate::uuid_util::parse_uuid(UUID).unwrap());
     if flow {

src/vless/vision_stream.rs

@@ -858,8 +858,11 @@ impl<S: VisionTransport> AsyncStream for VisionStream<S> {
 #[cfg(test)]
 mod tests {
     mod decisions;
+    mod native_decisions;
     mod native_errors;
     #[cfg(target_os = "linux")]
+    mod native_splice;
+    #[cfg(target_os = "linux")]
     mod splice;
 
     use super::*;

src/vless/vision_stream/tests/decisions.rs

@@ -5,9 +5,9 @@ use tokio::time::{Duration, timeout};
 
 type LegacyVision = VisionStream<CryptoTlsStream<DuplexStream>>;
 
-const APPLICATION: &[u8] = b"\x17\x03\x03\x00\x03app";
+pub(super) const APPLICATION: &[u8] = b"\x17\x03\x03\x00\x03app";
 const CHANGE_CIPHER_SPEC: &[u8] = b"\x14\x03\x03\x00\x01\x01";
-const TLS13_VERSION: &[u8] = b"\x00\x2b\x00\x02\x03\x04";
+pub(super) const TLS13_VERSION: &[u8] = b"\x00\x2b\x00\x02\x03\x04";
 
 fn record(content_type: u8, body: &[u8]) -> Vec<u8> {
     let mut bytes = vec![content_type, 3, 3];
@@ -16,14 +16,14 @@ fn record(content_type: u8, body: &[u8]) -> Vec<u8> {
     bytes
 }
 
-fn client_hello() -> Vec<u8> {
+pub(super) fn client_hello() -> Vec<u8> {
     let (mut client, _) = new_connections(Backend::Tls13);
     let mut flight = Vec::new();
     client.write_tls(&mut flight).unwrap();
     flight
 }
 
-fn server_hello(cipher: u16, session_id_len: u8, extensions: &[u8]) -> Vec<u8> {
+pub(super) fn server_hello(cipher: u16, session_id_len: u8, extensions: &[u8]) -> Vec<u8> {
     let mut body = vec![3, 3];
     body.extend_from_slice(&[42; 32]);
     body.push(session_id_len);
@@ -69,7 +69,11 @@ async fn pair(backend: Backend) -> (LegacyVision, LegacyVision) {
     )
 }
 
-async fn transfer(sender: &mut LegacyVision, receiver: &mut LegacyVision, data: &[u8]) {
+pub(super) async fn transfer<S: VisionTransport>(
+    sender: &mut VisionStream<S>,
+    receiver: &mut VisionStream<S>,
+    data: &[u8],
+) {
     timeout(Duration::from_secs(2), async {
         sender.write_all(data).await.unwrap();
         sender.flush().await.unwrap();
@@ -100,9 +104,9 @@ fn assert_terminal_write<S: VisionTransport>(stream: &VisionStream<S>, direct: b
     }
 }
 
-async fn assert_decision(
-    client: &mut LegacyVision,
-    server: &mut LegacyVision,
+pub(super) async fn assert_decision<S: VisionTransport>(
+    client: &mut VisionStream<S>,
+    server: &mut VisionStream<S>,
     direct: bool,
     server_first: bool,
 ) {

src/vless/vision_stream/tests/native_decisions.rs

@@ -0,0 +1,179 @@
+use super::decisions::{
+    APPLICATION, TLS13_VERSION, assert_decision, client_hello, server_hello, transfer,
+};
+use super::*;
+use crate::vless::encryption::{
+    ClientEncryption, EncryptionStream, ServerEncryption, tests::fixture_pair,
+};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt, DuplexStream},
+    time::{Duration, timeout},
+};
+
+type NativeVision = VisionStream<EncryptionStream<DuplexStream>>;
+
+async fn pair(
+    client: &ClientEncryption,
+    server: &ServerEncryption,
+    resumed: bool,
+) -> (NativeVision, NativeVision) {
+    let (a, b) = tokio::io::duplex(65536);
+    let (a, b) = timeout(Duration::from_secs(3), async {
+        tokio::join!(
+            async {
+                let mut stream = client.handshake(a).await?;
+                stream.flush().await?;
+                Ok::<_, io::Error>(stream)
+            },
+            server.handshake(b)
+        )
+    })
+    .await
+    .unwrap();
+    let a = a.unwrap();
+    assert_eq!(a.is_resumed(), resumed);
+    (
+        VisionStream::new_client(a, [7; 16])
+            .unwrap()
+            .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted),
+        VisionStream::new_server(b.unwrap(), [7; 16], b"")
+            .unwrap()
+            .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted),
+    )
+}
+
+#[tokio::test]
+async fn native_fresh_and_resumed_direct_decisions_preserve_both_directions() {
+    timeout(Duration::from_secs(60), async {
+        for mode in ["native", "xorpub", "random"] {
+            for cipher in [
+                0x1301, 0x1302, 0x1303, 0x1304, 0x1305, 0x1306, 0xc02f, 0x0a0a,
+            ] {
+                for extensions in [TLS13_VERSION, b"".as_slice()] {
+                    for server_first in [false, true] {
+                        let (client, server) = fixture_pair(mode, true, &[32]);
+                        for resumed in [false, true] {
+                            let (mut c, mut s) = pair(&client, &server, resumed).await;
+                            transfer(&mut c, &mut s, &client_hello()).await;
+                            transfer(&mut s, &mut c, &server_hello(cipher, 32, extensions)).await;
+                            assert_decision(
+                                &mut c,
+                                &mut s,
+                                (0x1301..=0x1304).contains(&cipher) && !extensions.is_empty(),
+                                server_first,
+                            )
+                            .await;
+                            c.shutdown().await.unwrap();
+                            assert_eq!(s.read(&mut [0]).await.unwrap(), 0);
+                            s.write_all(b"after half-close").await.unwrap();
+                            s.flush().await.unwrap();
+                            let mut reply = [0; 16];
+                            c.read_exact(&mut reply).await.unwrap();
+                            assert_eq!(&reply, b"after half-close");
+                            s.shutdown().await.unwrap();
+                            assert_eq!(c.read(&mut [0]).await.unwrap(), 0);
+                        }
+                    }
+                }
+            }
+        }
+    })
+    .await
+    .unwrap();
+}
+
+#[tokio::test]
+async fn native_fragmented_records_preserve_direct_decisions_and_coalesced_tails() {
+    timeout(Duration::from_secs(60), async {
+        for mode in ["native", "xorpub", "random"] {
+            for cipher in [0x1301, 0x1305] {
+                let hello = server_hello(cipher, 32, TLS13_VERSION);
+                for split in 1..hello.len() {
+                    let (client, server) = fixture_pair(mode, true, &[32]);
+                    for resumed in [false, true] {
+                        let (mut c, mut s) = pair(&client, &server, resumed).await;
+                        transfer(&mut c, &mut s, &client_hello()).await;
+                        s.write_all(&hello[..split]).await.unwrap();
+                        s.flush().await.unwrap();
+                        assert!(!s.filter.supports_xtls());
+                        s.write_all(&hello[split..]).await.unwrap();
+                        s.flush().await.unwrap();
+                        let mut received = vec![0; hello.len()];
+                        c.read_exact(&mut received).await.unwrap();
+                        assert_eq!(received, hello);
+                        assert_decision(&mut c, &mut s, cipher == 0x1301, split % 2 == 0).await;
+                    }
+                }
+                for split in 1..APPLICATION.len() {
+                    let (client, server) = fixture_pair(mode, true, &[32]);
+                    let (mut c, mut s) = pair(&client, &server, false).await;
+                    transfer(&mut c, &mut s, &client_hello()).await;
+                    transfer(&mut s, &mut c, &hello).await;
+                    s.write_all(&APPLICATION[..split]).await.unwrap();
+                    s.flush().await.unwrap();
+                    let tail = [APPLICATION[split..].to_vec(), b"coalesced tail".to_vec()].concat();
+                    s.write_all(&tail).await.unwrap();
+                    s.flush().await.unwrap();
+                    let expected = [APPLICATION, b"coalesced tail"].concat();
+                    let mut received = vec![0; expected.len()];
+                    c.read_exact(&mut received).await.unwrap();
+                    assert_eq!(received, expected);
+                    assert_eq!(
+                        c.read_mode,
+                        if cipher == 0x1301 {
+                            VisionMode::Direct
+                        } else {
+                            VisionMode::Tls
+                        }
+                    );
+                }
+            }
+        }
+    })
+    .await
+    .unwrap();
+}
+
+#[tokio::test]
+async fn native_malformed_hello_and_non_tls_never_enable_direct() {
+    timeout(Duration::from_secs(60), async {
+        let mut short = server_hello(0x1301, 32, TLS13_VERSION);
+        short[6..9].copy_from_slice(&[0, 0, 2]);
+        let mut long = short.clone();
+        long[6..9].copy_from_slice(&[0, 1, 0]);
+        let cases = [
+            short,
+            long,
+            server_hello(0x1301, 33, TLS13_VERSION),
+            server_hello(0x1301, 0, b"\x00\x2b\x00\x01\x03"),
+        ];
+        for mode in ["native", "xorpub", "random"] {
+            for hello in &cases {
+                let (client, server) = fixture_pair(mode, true, &[32]);
+                for resumed in [false, true] {
+                    let (mut c, mut s) = pair(&client, &server, resumed).await;
+                    transfer(&mut c, &mut s, &client_hello()).await;
+                    transfer(&mut s, &mut c, hello).await;
+                    assert_decision(&mut c, &mut s, false, false).await;
+                }
+            }
+            let (client, server) = fixture_pair(mode, true, &[32]);
+            for resumed in [false, true] {
+                let (mut c, mut s) = pair(&client, &server, resumed).await;
+                transfer(&mut c, &mut s, b"GET / HTTP/1.1\r\n\r\n").await;
+                transfer(&mut s, &mut c, b"HTTP/1.1 200 OK\r\n\r\n").await;
+                // Exhaust the shared read/write inspection budget, then commit END.
+                for _ in 0..10 {
+                    transfer(&mut c, &mut s, &[b'x'; 256]).await;
+                    transfer(&mut s, &mut c, &[b'y'; 256]).await;
+                }
+                assert_eq!(c.write_mode, VisionMode::Tls);
+                assert_eq!(s.read_mode, VisionMode::Tls);
+                assert!(!c.filter.supports_xtls());
+                assert!(!s.filter.supports_xtls());
+            }
+        }
+    })
+    .await
+    .unwrap();
+}

src/vless/vision_stream/tests/native_splice.rs

@@ -0,0 +1,125 @@
+use super::decisions::{APPLICATION, TLS13_VERSION, client_hello, server_hello};
+use super::*;
+use crate::{copy_bidirectional::copy_bidirectional, vless::encryption::tests::fixture_pair};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::{TcpListener, TcpStream},
+    time::{Duration, timeout},
+};
+
+async fn tcp_pair() -> (TcpStream, TcpStream) {
+    let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
+    let (client, server) = tokio::join!(
+        TcpStream::connect(("127.0.0.1", listener.local_addr().unwrap().port())),
+        listener.accept()
+    );
+    let client = client.unwrap();
+    let server = server.unwrap().0;
+    client.set_nodelay(true).unwrap();
+    server.set_nodelay(true).unwrap();
+    (client, server)
+}
+
+async fn transfer(
+    sender: &mut (impl AsyncStream + ?Sized),
+    receiver: &mut (impl AsyncStream + ?Sized),
+    bytes: &[u8],
+) {
+    let mut received = vec![0; bytes.len()];
+    let (sent, read) = tokio::join!(
+        async {
+            sender.write_all(bytes).await?;
+            sender.flush().await
+        },
+        receiver.read_exact(&mut received)
+    );
+    sent.unwrap();
+    read.unwrap();
+    assert_eq!(received, bytes);
+}
+
+#[tokio::test]
+async fn native_direct_splices_only_unmasked_tcp_and_preserves_half_close() {
+    timeout(Duration::from_secs(60), async {
+        for mode in ["native", "xorpub", "random"] {
+            for server_relay in [false, true] {
+                for hide_tcp in [false, true] {
+                    let (client, server) = fixture_pair(mode, true, &[32]);
+                    for resumed in [false, true] {
+                        let (a, b) = tcp_pair().await;
+                        let wrap = |socket| -> Box<dyn AsyncStream> {
+                            if hide_tcp {
+                                Box::new(crate::prepend_stream::PrependStream::new(socket, None))
+                            } else {
+                                Box::new(socket)
+                            }
+                        };
+                        let (a, b) = tokio::join!(
+                            async {
+                                let mut a = client.handshake(wrap(a)).await.unwrap();
+                                assert_eq!(a.is_resumed(), resumed);
+                                a.flush().await.unwrap();
+                                a
+                            },
+                            server.handshake(wrap(b))
+                        );
+                        let c = VisionStream::new_client(a, [7; 16])
+                            .unwrap()
+                            .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted);
+                        let s = VisionStream::new_server(b.unwrap(), [7; 16], b"")
+                            .unwrap()
+                            .with_write_error_policy(VisionWriteErrorPolicy::AcknowledgeAccepted);
+                        let (mut peer, mut relay_io) = if server_relay { (c, s) } else { (s, c) };
+                        let (mut remote, mut target) = tcp_pair().await;
+                        let relay = tokio::spawn(async move {
+                            copy_bidirectional(&mut relay_io, &mut remote, false, false).await
+                        });
+                        let hello = client_hello();
+                        let flight = [
+                            server_hello(0x1301, 32, TLS13_VERSION),
+                            APPLICATION.to_vec(),
+                        ]
+                        .concat();
+                        if server_relay {
+                            transfer(&mut peer, &mut target, &hello).await;
+                            transfer(&mut target, &mut peer, &flight).await;
+                            assert!(peer.plain_tcp().is_none());
+                            transfer(&mut peer, &mut target, APPLICATION).await;
+                        } else {
+                            transfer(&mut target, &mut peer, &hello).await;
+                            transfer(&mut peer, &mut target, &flight).await;
+                            assert!(peer.plain_tcp().is_none());
+                            transfer(&mut target, &mut peer, APPLICATION).await;
+                        }
+                        let expect_splice = mode != "random" && !hide_tcp;
+                        assert_eq!(peer.plain_tcp().is_some(), expect_splice);
+                        let payload: Vec<_> =
+                            (0..1024 * 1024 + 37).map(|n| (n % 251) as u8).collect();
+                        for upload in [true, false] {
+                            let before = crate::splice::TEST_SPLICED_BYTES.get();
+                            if upload {
+                                transfer(&mut peer, &mut target, &payload).await;
+                            } else {
+                                transfer(&mut target, &mut peer, &payload).await;
+                            }
+                            let copied = crate::splice::TEST_SPLICED_BYTES.get() - before;
+                            assert_eq!(
+                                copied > 0,
+                                expect_splice,
+                                "{mode}, server={server_relay}, hidden={hide_tcp}, resumed={resumed}, upload={upload}"
+                            );
+                        }
+                        peer.shutdown().await.unwrap();
+                        assert_eq!(target.read(&mut [0]).await.unwrap(), 0);
+                        transfer(&mut target, &mut peer, b"response after EOF").await;
+                        target.shutdown().await.unwrap();
+                        assert_eq!(peer.read(&mut [0]).await.unwrap(), 0);
+                        relay.await.unwrap().unwrap();
+                    }
+                }
+            }
+        }
+    })
+    .await
+    .unwrap();
+}

src/xhttp/carrier.rs

@@ -350,6 +350,7 @@ impl Carrier {
             Sender::H2(sender) => {
                 let (parts, body) = request.into_parts();
                 let empty = hyper::body::Body::is_end_stream(&body);
+                let streaming = matches!(&body, Body::Stream { .. });
                 let mut sender = tokio::time::timeout(timeouts.headers, sender.clone().ready())
                     .await?
                     .map_err(io::Error::other)?;
@@ -365,10 +366,16 @@ impl Carrier {
                         let _carrier = carrier;
                         let result = tokio::select! {
                             biased;
-                            _ = request_cancel.cancelled() => Err(io::ErrorKind::ConnectionAborted.into()),
+                            _ = request_cancel.cancelled() => Err(H2SendError::Local(io::ErrorKind::ConnectionAborted.into())),
                             result = send_h2_tracked(body, &mut send, upload) => result,
                         };
                         if let Err(error) = result {
+                            // RFC 9113 section 8.1 allows an early response followed by NO_ERROR.
+                            // Drain the response; its END_STREAM, not this reset, establishes EOF.
+                            if streaming && error.peer_sent_no_error(&mut send) {
+                                return;
+                            }
+                            let error = error.into_io();
                             log::debug!("XHTTP H2 upload failed: {error}");
                             send.send_reset(h2::Reason::CANCEL);
                             cancel.cancel();
@@ -551,41 +558,70 @@ pub async fn receive_h2(mut body: h2::RecvStream, mut writer: Writer) {
 }
 
 pub async fn send_h2(body: Body, mut send: h2::SendStream<Bytes>) -> io::Result<()> {
-    send_h2_tracked(body, &mut send, None).await
+    send_h2_tracked(body, &mut send, None)
+        .await
+        .map_err(H2SendError::into_io)
+}
+
+#[derive(Debug)]
+enum H2SendError {
+    Local(io::Error),
+    Stream(io::Error),
+}
+
+impl H2SendError {
+    fn into_io(self) -> io::Error {
+        match self {
+            Self::Local(error) | Self::Stream(error) => error,
+        }
+    }
+
+    fn peer_sent_no_error(&self, send: &mut h2::SendStream<Bytes>) -> bool {
+        // poll_capacity can return None and send_data can return a local stream-state error.
+        // Consult the stream's reason without hiding errors from the body producer.
+        matches!(self, Self::Stream(_))
+            && matches!(
+                send.poll_reset(&mut std::task::Context::from_waker(std::task::Waker::noop())),
+                std::task::Poll::Ready(Ok(h2::Reason::NO_ERROR))
+            )
+    }
+}
+
+impl From<h2::Error> for H2SendError {
+    fn from(error: h2::Error) -> Self {
+        Self::Stream(io::Error::other(error))
+    }
 }
 
 async fn send_h2_tracked(
     mut body: Body,
     send: &mut h2::SendStream<Bytes>,
     mut upload: Option<Upload>,
-) -> io::Result<()> {
+) -> Result<(), H2SendError> {
     'frames: loop {
         let frame = tokio::select! {
             frame = body.frame() => frame,
-            reset = futures::future::poll_fn(|cx| send.poll_reset(cx)) => return Err(io::Error::other(format!("XHTTP H2 stream reset: {reset:?}"))),
+            reset = futures::future::poll_fn(|cx| send.poll_reset(cx)) => return Err(H2SendError::Stream(io::Error::other(format!("XHTTP H2 stream reset: {reset:?}")))),
         };
         let Some(frame) = frame else {
-            send.send_data(Bytes::new(), true)
-                .map_err(io::Error::other)?;
+            send.send_data(Bytes::new(), true)?;
             break;
         };
-        let Ok(mut data) = frame?.into_data() else {
+        let Ok(mut data) = frame.map_err(H2SendError::Local)?.into_data() else {
             continue;
         };
         while !data.is_empty() {
             send.reserve_capacity(data.len());
             let capacity = futures::future::poll_fn(|cx| send.poll_capacity(cx))
                 .await
-                .ok_or_else(|| io::Error::from(io::ErrorKind::BrokenPipe))?
-                .map_err(io::Error::other)?;
+                .ok_or_else(|| H2SendError::Stream(io::ErrorKind::BrokenPipe.into()))??;
             let len = capacity.min(data.len());
             if len == 0 {
                 continue;
             }
             // Known-length peers may respond and reset immediately after the final bytes.
             let end = len == data.len() && hyper::body::Body::is_end_stream(&body);
-            send.send_data(data.split_to(len), end)
-                .map_err(io::Error::other)?;
+            send.send_data(data.split_to(len), end)?;
             if let Some(upload) = &upload {
                 upload.activity.touch();
             }
@@ -603,6 +639,344 @@ async fn send_h2_tracked(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::resources::Budget;
+    use tokio::io::AsyncReadExt;
+
+    async fn h2_peer() -> (
+        Arc<Carrier>,
+        tokio::sync::mpsc::UnboundedReceiver<(
+            Request<h2::RecvStream>,
+            h2::server::SendResponse<Bytes>,
+        )>,
+        Tasks,
+    ) {
+        let (client, peer) = tokio::io::duplex(4096);
+        let tasks = Tasks::new();
+        let (requests, received) = tokio::sync::mpsc::unbounded_channel();
+        tasks.spawn(async move {
+            let mut connection = h2::server::Builder::new()
+                .initial_window_size(1024)
+                .handshake(peer)
+                .await
+                .unwrap();
+            while let Some(request) = connection.accept().await {
+                if requests.send(request.unwrap()).is_err() {
+                    break;
+                }
+            }
+        });
+        let carrier = Carrier::tcp(Box::new(client), XhttpVersion::H2, 8192)
+            .await
+            .unwrap();
+        (carrier, received, tasks)
+    }
+
+    #[tokio::test]
+    async fn h2_early_response_drains_after_no_error_reset() {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            for blocked_upload in [false, true] {
+                for response_len in [0, 48 * 1024] {
+                    let (carrier, mut requests, _server) = h2_peer().await;
+                    let tasks = Tasks::new();
+                    let budget = Arc::new(Budget::new(Some(16 * 1024)));
+                    let memory = Memory::new(budget.clone(), 16 * 1024);
+                    let (mut upload, body) = pipe(memory.clone());
+                    let data = vec![7; if blocked_upload { 2048 } else { 1024 }];
+                    upload.write_all(&data).await.unwrap();
+                    let expected: Vec<_> = (0..response_len).map(|i| (i % 251) as u8).collect();
+                    let request = Request::post("http://localhost/stream-one")
+                        .body(Body::stream(body))
+                        .unwrap();
+                    let (response, ()) = tokio::join!(
+                        carrier.request(request, memory.clone(), &tasks, Duration::from_secs(1)),
+                        async {
+                            let (request, mut respond) = requests.recv().await.unwrap();
+                            let mut incoming = request.into_body();
+                            assert_eq!(incoming.data().await.unwrap().unwrap(), &data[..1024]);
+                            // Leave the upload open, including its exhausted window in the blocked case.
+                            let mut send = respond
+                                .send_response(Response::new(()), expected.is_empty())
+                                .unwrap();
+                            if !expected.is_empty() {
+                                send.send_data(Bytes::copy_from_slice(&expected), true)
+                                    .unwrap();
+                            }
+                            // h2 queues NO_ERROR after the complete response when these handles drop.
+                        }
+                    );
+                    let mut response = response.unwrap().into_body();
+                    upload.closed().await;
+                    assert!(upload.write_all(b"after reset").await.is_err());
+                    let mut actual = Vec::new();
+                    response.read_to_end(&mut actual).await.unwrap();
+                    assert_eq!(actual, expected);
+                    tasks.wait().await;
+                    assert!(!tasks.cancel.is_cancelled());
+                    drop(response);
+                    drop(upload);
+                    assert_eq!(budget.snapshot().active, 0);
+
+                    // The reset belongs to one stream, not the pooled connection.
+                    let probe_tasks = Tasks::new();
+                    let (probe, ()) = tokio::join!(
+                        carrier.request(
+                            Request::get("http://localhost/probe")
+                                .body(Body::Empty)
+                                .unwrap(),
+                            memory,
+                            &probe_tasks,
+                            Duration::from_secs(1),
+                        ),
+                        async {
+                            let (_, mut respond) = requests.recv().await.unwrap();
+                            let mut send = respond.send_response(Response::new(()), false).unwrap();
+                            send.send_data(Bytes::from_static(b"reused"), true).unwrap();
+                        }
+                    );
+                    let mut actual = Vec::new();
+                    probe
+                        .unwrap()
+                        .into_body()
+                        .read_to_end(&mut actual)
+                        .await
+                        .unwrap();
+                    assert_eq!(actual, b"reused");
+                    probe_tasks.wait().await;
+                    assert_eq!(budget.snapshot().active, 0);
+                }
+            }
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn h2_upload_failures_and_truncated_responses_are_not_eof() {
+        enum Failure {
+            Reset(h2::Reason),
+            Body,
+            Cancel,
+            Disconnect,
+        }
+        tokio::time::timeout(Duration::from_secs(5), async {
+            for failure in [
+                Failure::Reset(h2::Reason::NO_ERROR),
+                Failure::Reset(h2::Reason::CANCEL),
+                Failure::Reset(h2::Reason::INTERNAL_ERROR),
+                Failure::Body,
+                Failure::Cancel,
+                Failure::Disconnect,
+            ] {
+                let (carrier, mut requests, server) = h2_peer().await;
+                let tasks = Tasks::new();
+                let memory = Memory::new(Arc::new(Budget::new(Some(16 * 1024))), 16 * 1024);
+                let (mut upload, body) = pipe(memory.clone());
+                let request = Request::post("http://localhost/stream-one")
+                    .body(Body::stream(body))
+                    .unwrap();
+                let (response, mut send) = tokio::join!(
+                    carrier.request(request, memory, &tasks, Duration::from_secs(1)),
+                    async {
+                        let (_request, mut respond) = requests.recv().await.unwrap();
+                        let mut send = respond.send_response(Response::new(()), false).unwrap();
+                        send.send_data(Bytes::from_static(b"prefix"), false)
+                            .unwrap();
+                        send
+                    }
+                );
+                let mut response = response.unwrap().into_body();
+                let mut prefix = [0; 6];
+                response.read_exact(&mut prefix).await.unwrap();
+                assert_eq!(&prefix, b"prefix");
+                match failure {
+                    Failure::Reset(reason) => send.send_reset(reason),
+                    Failure::Body => upload.fail(io::ErrorKind::UnexpectedEof.into()).await,
+                    Failure::Cancel => tasks.cancel.cancel(),
+                    Failure::Disconnect => server.cancel.cancel(),
+                }
+                upload.closed().await;
+                assert!(response.read(&mut [0; 1]).await.is_err());
+                tasks.wait().await;
+                let quiet = matches!(failure, Failure::Reset(h2::Reason::NO_ERROR));
+                assert_eq!(tasks.cancel.is_cancelled(), !quiet);
+            }
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn h2_no_error_goaway_does_not_complete_a_truncated_response() {
+        use bytes::{BufMut, BytesMut};
+
+        fn frame(kind: u8, flags: u8, stream: u32, payload: &[u8]) -> Bytes {
+            let mut frame = BytesMut::new();
+            frame.put_uint(payload.len() as u64, 3);
+            frame.put_u8(kind);
+            frame.put_u8(flags);
+            frame.put_u32(stream);
+            frame.extend_from_slice(payload);
+            frame.freeze()
+        }
+
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let (client, mut peer) = tokio::io::duplex(4096);
+            let server = Tasks::new();
+            let shutdown = CancellationToken::new();
+            let stop = shutdown.clone();
+            let release_peer = CancellationToken::new();
+            let release = release_peer.clone();
+            server.spawn(async move {
+                let mut preface = [0; 24];
+                peer.read_exact(&mut preface).await.unwrap();
+                assert_eq!(&preface, b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n");
+                peer.write_all(&frame(4, 0, 0, &[])).await.unwrap();
+                loop {
+                    let mut header = [0; 9];
+                    peer.read_exact(&mut header).await.unwrap();
+                    let len = u32::from_be_bytes([0, header[0], header[1], header[2]]) as usize;
+                    let mut payload = vec![0; len];
+                    peer.read_exact(&mut payload).await.unwrap();
+                    match (header[3], header[4] & 1) {
+                        // Acknowledge SETTINGS and the carrier's readiness PING.
+                        (4, 0) => peer.write_all(&frame(4, 1, 0, &[])).await.unwrap(),
+                        (6, 0) => peer.write_all(&frame(6, 1, 0, &payload)).await.unwrap(),
+                        (1, _) => {
+                            assert_eq!(&header[5..], &1_u32.to_be_bytes());
+                            break;
+                        }
+                        _ => {}
+                    }
+                }
+                // HEADERS (:status 200) and DATA without END_STREAM.
+                peer.write_all(&frame(1, 4, 1, &[0x88])).await.unwrap();
+                peer.write_all(&frame(0, 0, 1, b"prefix")).await.unwrap();
+                stop.cancelled().await;
+                // Exclude stream 1 from GOAWAY so it receives NO_ERROR, not transport EOF.
+                peer.write_all(&frame(7, 0, 0, &[0; 8])).await.unwrap();
+                release.cancelled().await;
+                drop(peer);
+            });
+            let carrier = Carrier::tcp(Box::new(client), XhttpVersion::H2, 8192)
+                .await
+                .unwrap();
+            let tasks = Tasks::new();
+            let memory = Memory::new(Arc::new(Budget::new(Some(16 * 1024))), 16 * 1024);
+            let (upload, body) = pipe(memory.clone());
+            let mut response = carrier
+                .request(
+                    Request::post("http://localhost/stream-one")
+                        .body(Body::stream(body))
+                        .unwrap(),
+                    memory,
+                    &tasks,
+                    Duration::from_secs(1),
+                )
+                .await
+                .unwrap()
+                .into_body();
+            let mut prefix = [0; 6];
+            response.read_exact(&mut prefix).await.unwrap();
+            assert_eq!(&prefix, b"prefix");
+            shutdown.cancel();
+            let error = response.read(&mut [0; 1]).await.unwrap_err();
+            let error = error
+                .get_ref()
+                .unwrap()
+                .downcast_ref::<h2::Error>()
+                .unwrap();
+            assert!(error.is_go_away());
+            assert_eq!(error.reason(), Some(h2::Reason::NO_ERROR));
+            upload.closed().await;
+            tasks.wait().await;
+            assert!(!tasks.cancel.is_cancelled());
+            release_peer.cancel();
+            server.wait().await;
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn h2_early_response_does_not_acknowledge_incomplete_packet() {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let (carrier, mut requests, _server) = h2_peer().await;
+            let tasks = Tasks::new();
+            let budget = Arc::new(Budget::new(Some(16 * 1024)));
+            let memory = Memory::new(budget.clone(), 16 * 1024);
+            let (signal, written) = UploadSignal::channel();
+            let request = Request::post("http://localhost/packet/0")
+                .extension(signal)
+                .body(Body::Chunks(vec![memory.copy(&[7; 2048]).unwrap()].into()))
+                .unwrap();
+            let reservation = carrier.reserve().await.unwrap();
+            let (response, ()) = tokio::join!(
+                carrier.request_reserved(
+                    request,
+                    memory,
+                    &tasks,
+                    Timeouts::packet(Duration::from_secs(1), Duration::from_secs(1)),
+                    reservation,
+                ),
+                async {
+                    let (request, mut respond) = requests.recv().await.unwrap();
+                    let mut incoming = request.into_body();
+                    assert_eq!(incoming.data().await.unwrap().unwrap().len(), 1024);
+                    respond.send_response(Response::new(()), true).unwrap();
+                }
+            );
+            assert!(response.is_err());
+            assert!(written.await.is_err());
+            tasks.wait().await;
+            assert!(tasks.cancel.is_cancelled());
+            assert_eq!(budget.snapshot().active, 0);
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn h2_no_error_does_not_hide_body_errors_or_finish_server_sends() {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let (carrier, mut requests, _server) = h2_peer().await;
+            let Sender::H2(sender) = &carrier.sender else {
+                unreachable!()
+            };
+            let mut sender = sender.clone().ready().await.unwrap();
+            let (response, mut send) = sender
+                .send_request(
+                    Request::post("http://localhost/stream-one")
+                        .body(())
+                        .unwrap(),
+                    false,
+                )
+                .unwrap();
+            let (request, mut respond) = requests.recv().await.unwrap();
+            respond.send_response(Response::new(()), true).unwrap();
+            drop((request, respond));
+            let reason = futures::future::poll_fn(|cx| send.poll_reset(cx))
+                .await
+                .unwrap();
+            assert_eq!(reason, h2::Reason::NO_ERROR);
+
+            assert!(
+                futures::future::poll_fn(|cx| send.poll_capacity(cx))
+                    .await
+                    .is_none()
+            );
+            let error = H2SendError::Stream(io::ErrorKind::BrokenPipe.into());
+            assert!(error.peer_sent_no_error(&mut send));
+            // A reset can arrive between the body EOF poll and sending the final DATA frame.
+            let error = H2SendError::from(send.send_data(Bytes::new(), true).unwrap_err());
+            assert!(error.peer_sent_no_error(&mut send));
+            let error = H2SendError::Local(io::ErrorKind::UnexpectedEof.into());
+            assert!(!error.peer_sent_no_error(&mut send));
+            assert!(send_h2(Body::Empty, send).await.is_err());
+            assert!(response.await.unwrap().into_body().data().await.is_none());
+        })
+        .await
+        .unwrap();
+    }
 
     #[tokio::test(start_paused = true)]
     async fn packet_response_wait_requires_upload_completion_and_observes_cancellation() {

src/xhttp/lifecycle_tests.rs

@@ -211,6 +211,73 @@ async fn delayed_headers_half_close_and_pool_isolation() {
     }
 }
 
+#[tokio::test]
+async fn h2_streaming_early_responses_preserve_download_tails() {
+    for mode in [XhttpMode::StreamOne, XhttpMode::StreamUp] {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let listener = tokio::net::TcpListener::bind("0.0.0.0:0").await.unwrap();
+            let address = ([127, 0, 0, 1], listener.local_addr().unwrap().port()).into();
+            let tasks = Tasks::new();
+            let expected: Vec<_> = (0..96 * 1024).map(|i| (i % 251) as u8).collect();
+            let response_bytes = expected.clone();
+            tasks.spawn(async move {
+                let (socket, _) = listener.accept().await.unwrap();
+                let mut connection = h2::server::Builder::new()
+                    .initial_window_size(1024)
+                    .handshake(socket)
+                    .await
+                    .unwrap();
+                let requests = Tasks::new();
+                let uploaded = tokio_util::sync::CancellationToken::new();
+                while let Some(Ok((request, mut respond))) = connection.accept().await {
+                    let uploaded = uploaded.clone();
+                    let response_bytes = response_bytes.clone();
+                    requests.spawn(async move {
+                        let post = request.method() == http::Method::POST;
+                        let mut incoming = request.into_body();
+                        if post {
+                            assert_eq!(incoming.data().await.unwrap().unwrap(), &[7; 1024][..]);
+                            uploaded.cancel();
+                        } else {
+                            uploaded.cancelled().await;
+                        }
+                        let upload_only = post && mode == XhttpMode::StreamUp;
+                        let mut send = respond
+                            .send_response(Response::new(()), upload_only)
+                            .unwrap();
+                        if !upload_only {
+                            send.send_data(Bytes::from(response_bytes), true).unwrap();
+                        }
+                        // Dropping the unread POST queues NO_ERROR after its response.
+                    });
+                }
+            });
+            let config = serde_json::from_value(serde_json::json!({
+                "limits":{"max_buffered_bytes":65536,"max_session_buffered_bytes":65536},
+                "protocol":{"type":"socks"}
+            }))
+            .unwrap();
+            let client = configured_client(config, XhttpVersion::H2, mode);
+            let dialer = Arc::new(Dialer {
+                address,
+                key: RouteKey(vec![1]),
+                opens: AtomicUsize::new(0),
+            });
+            let mut stream = client.open(dialer.clone()).await.unwrap();
+            stream.write_all(&[7; 2048]).await.unwrap();
+            stream.write.closed().await;
+            let mut actual = Vec::new();
+            stream.read_to_end(&mut actual).await.unwrap();
+            assert_eq!(actual, expected);
+            stream._tasks.wait().await;
+            assert!(stream.write_all(b"after response").await.is_err());
+            assert_eq!(dialer.opens.load(Ordering::Relaxed), 1);
+        })
+        .await
+        .unwrap();
+    }
+}
+
 #[tokio::test]
 async fn idle_routes_are_evicted_and_failed_dials_do_not_fill_the_pool() {
     let (address, _server) = echo_peer(XhttpVersion::H2, false).await;

src/xhttp/server.rs

@@ -375,15 +375,23 @@ mod tests {
                     state.close(suffix);
                 }
                 let mut data = Vec::new();
-                let _ =
+                let result =
                     tokio::time::timeout(Duration::from_secs(1), response.read_to_end(&mut data))
                         .await
                         .unwrap();
                 assert!(data.is_empty());
-                // H3's idle sender observes STOP_SENDING on its ten-second probe.
-                tokio::time::timeout(Duration::from_secs(12), upload.cancel.cancelled())
-                    .await
-                    .unwrap();
+                if version == XhttpVersion::H2 {
+                    result.unwrap();
+                    tokio::time::timeout(Duration::from_secs(1), upload.wait())
+                        .await
+                        .unwrap();
+                    assert!(!upload.cancel.is_cancelled());
+                } else {
+                    // H3's idle sender observes STOP_SENDING on its ten-second probe.
+                    tokio::time::timeout(Duration::from_secs(12), upload.cancel.cancelled())
+                        .await
+                        .unwrap();
+                }
                 let next = Tasks::new();
                 let response = tokio::time::timeout(
                     Duration::from_secs(2),

tests/XHTTP.md

@@ -3,7 +3,7 @@
 The `xhttp_interop` integration target runs real Shoes processes against Shoes,
 Xray-core, and Mihomo Alpha. CI builds and caches these peers from pinned source:
 
-- Xray-core: `701af60772cda123492f86e383e1fdba066614a9` (26.10.10).
+- Xray-core: `92ccc939aaf4f8965f369cf0b142e005aeaa11d5` (26.10.17).
 - Mihomo Alpha: `94b13aed49b8dc41973e4bb3f721c5fa2c5c5c1f`.
 
 Use Go 1.27.1, `GOMAXPROCS=1`, and `go build -p 1`. Build Xray's `./main` package.
@@ -113,10 +113,10 @@ This extended sweep is opt-in; normal CI runs the focused recovery/containment c
 
 ### Local Stall Qualification
 
-On 2026-10-16, the pinned Xray client completed all six 16 MiB duplex cases against
-Shoes with 32 KiB POSTs, 30 ms pacing, no added RTT, and destination pauses of
-10/30/60 seconds. These correctness runs used a debug Shoes binary; their rates
-are not production throughput measurements.
+On 2026-10-16, the then-pinned Xray 26.10.10 (`701af607`) client completed all six
+16 MiB duplex cases against Shoes with 32 KiB POSTs, 30 ms pacing, no added RTT,
+and destination pauses of 10/30/60 seconds. These correctness runs used a debug
+Shoes binary; their rates are not production throughput measurements.
 
 | Server request cap | Total duration, 10/30/60 s pause | Peak service handlers | Peak application bytes | Combined peer RSS range |
 | --- | --- | --- | --- | --- |
@@ -137,8 +137,9 @@ for Shoes-to-Xray. Each session resumed with exact upload and download bytes.
 
 ### Local Throughput Qualification
 
-The release Shoes binary passed all 24 combinations of 0/50/150/300 ms added RTT,
-32 KiB/1,000,000-byte POST limits, and the three peer directions. All used 30 ms
+Using the then-pinned Xray 26.10.10 (`701af607`), the release Shoes binary passed
+all 24 combinations of 0/50/150/300 ms added RTT, 32 KiB/1,000,000-byte POST limits,
+and the three peer directions. All used 30 ms
 pacing and exact 16 MiB transfers each way; Shoes used its default client pipeline
 depth of 30. Representative aggregate duplex rates for 1,000,000-byte POSTs:
 

tests/cases/optional_peer/tls_hybrid.rs

@@ -13,7 +13,7 @@ use tokio::{
 
 fn start_xray(config: serde_json::Value) -> io::Result<(ProcessGuard, tempfile::NamedTempFile)> {
     let binary = std::env::var_os("SHOES_TEST_XRAY_BIN").unwrap_or_else(|| "xray".into());
-    let mut file = tempfile::NamedTempFile::new()?;
+    let mut file = tempfile::Builder::new().suffix(".json").tempfile()?;
     file.write_all(config.to_string().as_bytes())?;
     file.flush()?;
     let child = Command::new(binary)

tests/cases/optional_peer/vless_encryption.rs

@@ -23,14 +23,21 @@ const LIMIT: Duration = Duration::from_secs(10);
 
 #[path = "vless_encryption/benchmarks.rs"]
 mod benchmarks;
+#[path = "vless_encryption/chained.rs"]
+mod chained;
+#[path = "vless_encryption/composition.rs"]
+mod composition;
 #[path = "vless_encryption/legacy.rs"]
 mod legacy;
+#[path = "vless_encryption/lifecycle.rs"]
+mod lifecycle;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum Kind {
     Shoes,
     Xray,
     Mihomo,
+    SingBox,
 }
 
 impl Kind {
@@ -45,6 +52,12 @@ impl Kind {
             Self::Mihomo => std::env::var_os("MIHOMO_BINARY")
                 .expect("MIHOMO_BINARY must name the pinned Mihomo binary; no silent skips")
                 .into(),
+            Self::SingBox => {
+                std::env::var_os("SHOES_TEST_SING_BOX_BIN").expect(
+                    "SHOES_TEST_SING_BOX_BIN must name the pinned sing-box binary; no silent skips",
+                );
+                shoes_test_support::process::find_singbox_binary().unwrap()
+            }
         }
     }
 }
@@ -70,6 +83,20 @@ impl Peer {
     fn logs(&self) -> String {
         std::fs::read_to_string(self.directory.path().join("peer.log")).unwrap_or_default()
     }
+
+    fn assert_handshakes(&self, expected: &[&str]) {
+        let logs = self.logs();
+        let actual: Vec<_> = logs
+            .lines()
+            .filter_map(|line| {
+                let (_, handshake) = line
+                    .split_once("VLESS encryption client: ")
+                    .or_else(|| line.split_once("VLESS encryption server: "))?;
+                handshake.strip_suffix(" handshake")
+            })
+            .collect();
+        assert_eq!(actual, expected, "{logs}");
+    }
 }
 
 fn pair_logs(client: &Peer, server: &Peer) -> String {
@@ -88,7 +115,17 @@ async fn start(kind: Kind, port: u16, config: Value) -> Peer {
     start_binary(kind, port, config, kind.binary()).await
 }
 
-async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf) -> Peer {
+async fn start_binary(kind: Kind, port: u16, config: Value, binary: PathBuf) -> Peer {
+    start_options(kind, port, config, binary, None).await
+}
+
+async fn start_options(
+    kind: Kind,
+    port: u16,
+    mut config: Value,
+    binary: PathBuf,
+    go_debug: Option<&str>,
+) -> Peer {
     // Mihomo opens listeners before its tunnel becomes Running. A separate
     // direct route probes readiness without warming any native ticket cache.
     let mut readiness = if kind == Kind::Mihomo {
@@ -118,15 +155,19 @@ async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf)
     let mut command = Command::new(binary);
     match kind {
         Kind::Shoes => {
-            command.args(["--threads", "1", "--no-reload"]).arg(&path);
+            command.args(["--threads", "1", "--no-reload"]);
+            command.arg(&path);
         }
-        Kind::Xray => {
+        Kind::Xray | Kind::SingBox => {
             command.args(["run", "-c"]).arg(&path);
         }
         Kind::Mihomo => {
             command.arg("-d").arg(directory.path()).arg("-f").arg(&path);
         }
     }
+    if let Some(value) = go_debug {
+        command.env("GODEBUG", value);
+    }
     let child = command
         .env("GOMAXPROCS", "1")
         .env("RUST_LOG", "shoes=debug")
@@ -175,12 +216,33 @@ async fn start_binary(kind: Kind, port: u16, mut config: Value, binary: PathBuf)
 }
 
 async fn free_port() -> u16 {
-    TcpListener::bind("0.0.0.0:0")
-        .await
-        .unwrap()
-        .local_addr()
-        .unwrap()
-        .port()
+    static ALLOCATED: Mutex<Vec<u16>> = Mutex::new(Vec::new());
+    for _ in 0..100 {
+        let tcp = TcpListener::bind("0.0.0.0:0").await.unwrap();
+        let port = tcp.local_addr().unwrap().port();
+        // SOCKS and QUIC peers also bind UDP on this port.
+        match tokio::net::UdpSocket::bind(("0.0.0.0", port)).await {
+            Ok(_udp) => {
+                let mut allocated = ALLOCATED.lock().unwrap();
+                if allocated.contains(&port) {
+                    continue;
+                }
+                allocated.push(port);
+                return port;
+            }
+            Err(error) if error.kind() == io::ErrorKind::AddrInUse => continue,
+            Err(error) => panic!("reserve UDP port: {error}"),
+        }
+    }
+    panic!("could not find a free TCP/UDP port");
+}
+
+#[tokio::test]
+async fn allocated_peer_ports_are_not_reused_before_listeners_start() {
+    let mut ports = std::collections::HashSet::new();
+    for _ in 0..512 {
+        assert!(ports.insert(free_port().await));
+    }
 }
 
 fn credentials(mode: &str, auth: &[&str]) -> (String, String, usize) {
@@ -231,6 +293,11 @@ fn server_config(kind: Kind, port: u16, decryption: &str, vision: bool) -> Value
             "listeners": [{"name": "encrypted", "type": "vless", "listen": "0.0.0.0", "port": port,
                 "decryption": decryption, "users": [{"username": "test", "uuid": USER, "flow": flow}]}],
         }),
+        Kind::SingBox => {
+            assert_eq!(decryption, "none");
+            json!({"log":{"level":"debug"},"inbounds":[{"type":"vless","listen":"0.0.0.0","listen_port":port,
+                "users":[{"uuid":USER,"flow":flow}]}],"outbounds":[{"type":"direct"}]})
+        }
     }
 }
 
@@ -255,6 +322,11 @@ fn client_config(kind: Kind, port: u16, server_port: u16, encryption: &str, visi
                 "uuid": USER, "encryption": encryption, "flow": flow, "udp": true, "packet-encoding": "packetaddr"}],
             "mode": "rule", "rules": ["MATCH,encrypted"],
         }),
+        Kind::SingBox => {
+            assert_eq!(encryption, "none");
+            json!({"log":{"level":"debug"},"inbounds":[{"type":"socks","listen":"0.0.0.0","listen_port":port}],
+                "outbounds":[{"type":"vless","server":"127.0.0.1","server_port":server_port,"uuid":USER,"flow":flow}]})
+        }
     }
 }
 
@@ -596,7 +668,13 @@ enum Transport {
     Raw,
     Tls,
     Websocket,
+    Wss,
     Xhttp,
+    XhttpMode {
+        version: &'static str,
+        mode: &'static str,
+    },
+    Quic,
     Reality,
 }
 
@@ -616,6 +694,21 @@ fn xray_endpoint_mut(config: &mut Value, server: bool) -> &mut Value {
     }
 }
 
+fn reality_keys() -> (String, String) {
+    use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+
+    let private = [42; 32];
+    let public =
+        aws_lc_rs::agreement::PrivateKey::from_private_key(&aws_lc_rs::agreement::X25519, &private)
+            .unwrap()
+            .compute_public_key()
+            .unwrap();
+    (
+        URL_SAFE_NO_PAD.encode(private),
+        URL_SAFE_NO_PAD.encode(public.as_ref()),
+    )
+}
+
 fn wrap(
     kind: Kind,
     mut config: Value,
@@ -626,6 +719,64 @@ fn wrap(
     cover_port: u16,
 ) -> Value {
     match (kind, transport) {
+        (Kind::SingBox, _) => {
+            return legacy::wrap_singbox(config, server, transport, cert, key, cover_port);
+        }
+        (_, Transport::Wss) => {
+            config = wrap(
+                kind,
+                config,
+                server,
+                Transport::Websocket,
+                cert,
+                key,
+                cover_port,
+            );
+            // TLS wraps the WebSocket stream, not the inner VLESS stream.
+            let ws = if kind == Kind::Xray {
+                Some(xray_endpoint_mut(&mut config, server)["streamSettings"].clone())
+            } else {
+                None
+            };
+            config = wrap(kind, config, server, Transport::Tls, cert, key, cover_port);
+            if let Some(ws) = ws {
+                let stream = &mut xray_endpoint_mut(&mut config, server)["streamSettings"];
+                stream["network"] = ws["network"].clone();
+                stream["wsSettings"] = ws["wsSettings"].clone();
+            }
+        }
+        (_, Transport::XhttpMode { version, mode }) => {
+            return composition::wrap_xhttp(
+                kind,
+                config,
+                server,
+                (version, mode),
+                cert,
+                key,
+                cover_port,
+            );
+        }
+        (Kind::Shoes, Transport::Quic) => {
+            let endpoint = if server {
+                &mut config[0]
+            } else {
+                &mut config[0]["rules"][0]["client_chain"]
+            };
+            endpoint["transport"] = json!("quic");
+            endpoint["quic_settings"] = if server {
+                json!({"cert": cert, "key": key})
+            } else {
+                json!({"verify": false, "sni_hostname": "localhost"})
+            };
+            if server {
+                let address = config[0]["address"].clone();
+                config
+                    .as_array_mut()
+                    .unwrap()
+                    .push(json!({"address":address,"protocol":{"type":"socks"}}));
+            }
+        }
+        (_, Transport::Quic) => unreachable!("raw QUIC is a Shoes transport"),
         (_, Transport::Raw) => {}
         (Kind::Shoes, Transport::Tls) => {
             let field = shoes_protocol_mut(&mut config, server);
@@ -694,17 +845,7 @@ fn wrap(
             field["streamSettings"] = json!({"network": "xhttp", "security": "none", "xhttpSettings": {"path": "/native", "host": "localhost", "mode": "packet-up"}});
         }
         (Kind::Shoes | Kind::Xray, Transport::Reality) => {
-            use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
-            let private = [42; 32];
-            let public = aws_lc_rs::agreement::PrivateKey::from_private_key(
-                &aws_lc_rs::agreement::X25519,
-                &private,
-            )
-            .unwrap()
-            .compute_public_key()
-            .unwrap();
-            let public = URL_SAFE_NO_PAD.encode(public.as_ref());
-            let private = URL_SAFE_NO_PAD.encode(private);
+            let (private, public) = reality_keys();
             if kind == Kind::Shoes {
                 let field = shoes_protocol_mut(&mut config, server);
                 let inner = field.take();
@@ -724,8 +865,31 @@ fn wrap(
                     json!({"network": "raw", "security": "reality", "realitySettings": reality});
             }
         }
-        (Kind::Mihomo, Transport::Xhttp | Transport::Reality) => {
-            unreachable!("targeted Xray-only transport cases")
+        (Kind::Mihomo, Transport::Xhttp) => {
+            return composition::wrap_xhttp(
+                kind,
+                config,
+                server,
+                ("plain", "packet-up"),
+                cert,
+                key,
+                cover_port,
+            );
+        }
+        (Kind::Mihomo, Transport::Reality) => {
+            let (private, public) = reality_keys();
+            if server {
+                config["listeners"][0]["reality-config"] = json!({
+                    "dest":format!("localhost:{cover_port}"), "server-names":["localhost"],
+                    "private-key":private, "short-id":["0123456789abcdef"]
+                });
+            } else {
+                let proxy = &mut config["proxies"][0];
+                proxy["tls"] = json!(true);
+                proxy["servername"] = json!("localhost");
+                proxy["client-fingerprint"] = json!("chrome");
+                proxy["reality-opts"] = json!({"public-key":public, "short-id":"0123456789abcdef"});
+            }
         }
     }
     config
@@ -797,6 +961,10 @@ async fn direct_case(peer: Kind, peer_client: bool, transport: Transport, mode:
                     );
                 }
             }
+            // WebSocket and Mihomo's masked native wrapper can close both transport halves.
+            tls.get_mut().1.send_close_notify();
+            tls.flush().await?;
+            assert_eq!(tls.read(&mut [0]).await?, 0, "data after TLS close");
         }
         Ok::<_, io::Error>(())
     })
@@ -805,14 +973,17 @@ async fn direct_case(peer: Kind, peer_client: bool, transport: Transport, mode:
         panic!("{result:?}\n{}", pair_logs(&client, &server));
     }
     let shoes = if peer_client { &server } else { &client };
-    assert_eq!(
-        shoes
-            .logs()
-            .contains("VISION WRITE: Switching to direct copy mode"),
-        !tls12,
-        "{}",
-        shoes.logs()
-    );
+    shoes.assert_handshakes(&["fresh", "resumed"]);
+    for direction in ["READ", "WRITE"] {
+        assert_eq!(
+            shoes.logs().contains(&format!(
+                "VISION {direction}: Switching to direct copy mode"
+            )),
+            !tls12,
+            "{}",
+            shoes.logs()
+        );
+    }
     drop((client, server));
     tap_task.stop().await;
     target_task.stop().await;
@@ -1014,6 +1185,9 @@ async fn wrong_uuid_and_key_order_have_no_destination_side_effects() {
                                 [0]["id"]
                         }
                         Kind::Mihomo => &mut config["proxies"][0]["uuid"],
+                        Kind::SingBox => {
+                            unreachable!("sing-box does not implement native encryption")
+                        }
                     };
                     *user = json!("c85798ef-e9dc-46a4-9a87-8da4499d36d0");
                 } else {

tests/cases/optional_peer/vless_encryption/chained.rs

@@ -0,0 +1,99 @@
+use super::*;
+
+#[tokio::test]
+async fn tls_and_anytls_consumers_survive_backpressure_and_cancelled_streams() {
+    let with_padding_gap = |setting: String| {
+        let gapped = setting.replace(".100-35-35.", ".100-35-35.100-25-25.100-35-35.");
+        assert_ne!(gapped, setting, "fixture padding token was not found");
+        gapped
+    };
+    for mode in ["native", "xorpub", "random"] {
+        for vision in [false, true] {
+            for anytls in [false, true] {
+                eprintln!("chained: {mode}, vision={vision}, anytls={anytls}");
+                let (encryption, decryption, _) = credentials(mode, &["x25519"]);
+                let encryption = with_padding_gap(encryption);
+                let decryption = with_padding_gap(decryption);
+                let carrier_port = free_port().await;
+                let carrier = start(
+                    Kind::Shoes,
+                    carrier_port,
+                    server_config(Kind::Shoes, carrier_port, &decryption, vision),
+                )
+                .await;
+                let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
+                let hop_port = free_port().await;
+                let hop_protocol = if anytls {
+                    json!({"type":"anytls","users":[{"name":"test","password":"test-password"}]})
+                } else {
+                    json!({"type":"socks"})
+                };
+                let hop_config = json!([{
+                    "address":format!("0.0.0.0:{hop_port}"),
+                    "protocol":{"type":"tls","default_target":{
+                        "cert":cert.cert.pem(),"key":cert.signing_key.serialize_pem(),"protocol":hop_protocol
+                    }}
+                }]);
+                let hop = start(Kind::Shoes, hop_port, hop_config).await;
+                let proxy = free_port().await;
+                let mut config =
+                    client_config(Kind::Shoes, proxy, carrier_port, &encryption, vision);
+                let vless_hop = config[0]["rules"][0]["client_chain"].take();
+                let hop_client_protocol = if anytls {
+                    json!({"type":"anytls","password":"test-password"})
+                } else {
+                    json!({"type":"socks"})
+                };
+                config[0]["rules"][0]["client_chain"] = json!([vless_hop, {
+                    "address":format!("127.0.0.1:{hop_port}"),
+                    "protocol":{"type":"tls","sni_hostname":"localhost","verify":false,"protocol":hop_client_protocol}
+                }]);
+                let client = start(Kind::Shoes, proxy, config).await;
+                let (target, _, mut target_task) = echo().await;
+                let result = timeout(Duration::from_secs(30), async {
+                    // Keep one stream alive while abandoning a different stream with unread data.
+                    let mut survivor = socks_connect(proxy, target).await?;
+                    let mut cancelled = socks_connect(proxy, target).await?;
+                    cancelled.write_all(&vec![31; 128 * 1024]).await?;
+                    drop(cancelled);
+                    let payload: Vec<_> =
+                        (0..2 * 1024 * 1024 + 17).map(|n| (n % 251) as u8).collect();
+                    let (mut read, mut write) = survivor.split();
+                    tokio::try_join!(
+                        async {
+                            for chunk in payload.chunks(16384) {
+                                write.write_all(chunk).await?;
+                            }
+                            write.flush().await
+                        },
+                        async {
+                            let mut received = vec![0; payload.len()];
+                            for chunk in received.chunks_mut(4096) {
+                                read.read_exact(chunk).await?;
+                                tokio::time::sleep(Duration::from_millis(1)).await;
+                            }
+                            assert_eq!(received, payload);
+                            Ok::<_, io::Error>(())
+                        }
+                    )?;
+                    drop(survivor);
+                    exchange(proxy, target, b"new stream after cancellation").await
+                })
+                .await;
+                assert!(
+                    matches!(result, Ok(Ok(()))),
+                    "{result:?}\n{}\nhop: {}",
+                    pair_logs(&client, &carrier),
+                    hop.logs()
+                );
+                assert!(
+                    carrier
+                        .logs()
+                        .contains("VLESS encryption server: fresh handshake")
+                );
+                drop((client, hop, carrier));
+                target_task.stop().await;
+            }
+        }
+    }
+}

tests/cases/optional_peer/vless_encryption/composition.rs

@@ -0,0 +1,309 @@
+use super::*;
+use shoes_test_support::{
+    socks5::Socks5UdpAssociation, test_servers::start_udp_echo_server_with_suffix,
+};
+
+pub(super) fn wrap_xhttp(
+    kind: Kind,
+    mut config: Value,
+    server: bool,
+    (version, mode): (&str, &str),
+    cert: &str,
+    key: &str,
+    cover: u16,
+) -> Value {
+    let carrier = match (version, kind) {
+        ("reality", _) => Some(Transport::Reality),
+        ("plain" | "h2c", _) => None,
+        ("h3", Kind::Shoes) => Some(Transport::Quic),
+        _ => Some(Transport::Tls),
+    };
+    if let Some(carrier) = carrier {
+        config = wrap(kind, config, server, carrier, cert, key, cover);
+    }
+    let alpn = match version {
+        "plain" | "h1" => "http/1.1",
+        "h3" => "h3",
+        _ => "h2",
+    };
+    match kind {
+        Kind::SingBox => unreachable!("sing-box does not implement XHTTP"),
+        Kind::Shoes => {
+            let mut protocol = shoes_protocol_mut(&mut config, server);
+            if protocol["type"] == "tls" {
+                protocol = if !server {
+                    &mut protocol["protocol"]
+                } else if version == "reality" {
+                    &mut protocol["reality_targets"]["localhost"]["protocol"]
+                } else {
+                    &mut protocol["default_target"]["protocol"]
+                };
+            } else if protocol["type"] == "reality" {
+                protocol = &mut protocol["protocol"];
+            }
+            let inner = protocol.take();
+            let http_version = match version {
+                "plain" => "h1",
+                "h2c" => "h2",
+                "reality" => "auto",
+                other => other,
+            };
+            *protocol = json!({"type":"xhttp", "path":"/native", "host":"localhost", "mode":mode,
+                "http_version":http_version,
+                "sc_max_each_post_bytes":1024, "protocol":inner});
+        }
+        Kind::Xray => {
+            let stream = &mut xray_endpoint_mut(&mut config, server)["streamSettings"];
+            if stream.is_null() {
+                *stream = json!({"security":"none"});
+            }
+            stream["network"] = json!("xhttp");
+            stream["xhttpSettings"] = json!({"path":"/native", "host":"localhost", "mode":mode,
+                "extra":{"scMaxEachPostBytes":1024, "scMinPostsIntervalMs":0}});
+            if stream["security"] == "tls" {
+                stream["tlsSettings"]["alpn"] = json!([alpn]);
+            }
+            if server && version == "h3" {
+                let port = config["inbounds"][0]["port"].clone();
+                config["inbounds"].as_array_mut().unwrap().push(json!({
+                    "listen":"0.0.0.0","port":port,"protocol":"socks","settings":{"auth":"noauth"}
+                }));
+            }
+        }
+        Kind::Mihomo => {
+            let options = json!({"path":"/native", "host":"localhost", "mode":mode,
+                "sc-max-each-post-bytes":"1024", "sc-min-posts-interval-ms":"1"});
+            if server {
+                assert_ne!(version, "h3", "Mihomo does not serve XHTTP over QUIC");
+                config["listeners"][0]["xhttp-config"] = options;
+                if matches!(version, "plain" | "h2c") {
+                    config["listeners"][0]["allow-insecure"] = json!(true);
+                }
+            } else {
+                config["proxies"][0]["network"] = json!("xhttp");
+                config["proxies"][0]["alpn"] = json!([alpn]);
+                config["proxies"][0]["xhttp-opts"] = options;
+            }
+        }
+    }
+    config
+}
+
+async fn wrapped_case(
+    peer: Kind,
+    peer_client: bool,
+    transport: Transport,
+    encrypted: bool,
+    vision: bool,
+) {
+    eprintln!(
+        "composition: {peer:?} client={peer_client} {transport:?} encrypted={encrypted} vision={vision}"
+    );
+    let (encryption, decryption) = if encrypted {
+        let (encryption, decryption, _) = credentials("random", &["mlkem768"]);
+        (encryption, decryption)
+    } else {
+        ("none".into(), "none".into())
+    };
+    let (server_kind, client_kind) = peer_roles(peer, peer_client);
+    let (cover, connector, mut cover_task) = tls_echo(false).await;
+    let (target, _, mut target_task) = echo().await;
+    let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
+    let (cert, key) = (cert.cert.pem(), cert.signing_key.serialize_pem());
+    let server_port = free_port().await;
+    let mut server_settings = server_config(server_kind, server_port, &decryption, vision);
+    if server_kind == Kind::Mihomo && !encrypted {
+        server_settings["listeners"][0]["allow-insecure"] = json!(true);
+    }
+    let server = start(
+        server_kind,
+        server_port,
+        wrap(
+            server_kind,
+            server_settings,
+            true,
+            transport,
+            &cert,
+            &key,
+            cover,
+        ),
+    )
+    .await;
+    let client_port = free_port().await;
+    let mut client_settings =
+        client_config(client_kind, client_port, server_port, &encryption, vision);
+    if client_kind == Kind::Mihomo {
+        client_settings["proxies"][0]["packet-encoding"] = json!("xudp");
+    }
+    let client = start(
+        client_kind,
+        client_port,
+        wrap(
+            client_kind,
+            client_settings,
+            false,
+            transport,
+            &cert,
+            &key,
+            cover,
+        ),
+    )
+    .await;
+    let data: Vec<_> = (0..32769).map(|n| (n % 251) as u8).collect();
+    for _ in 0..2 {
+        if vision {
+            let result = timeout(LIMIT, async {
+                let socket = socks_connect(client_port, cover).await?;
+                let mut stream = connector
+                    .connect("localhost".try_into().unwrap(), socket)
+                    .await?;
+                for payload in [&[42; 64][..], &[43; 64], &data] {
+                    let (mut r, mut w) = tokio::io::split(&mut stream);
+                    let mut reply = vec![0; payload.len()];
+                    tokio::try_join!(w.write_all(payload), r.read_exact(&mut reply))?;
+                    assert_eq!(reply, payload);
+                }
+                // Close inner TLS without requiring WebSocket to support a TCP half-close.
+                stream.get_mut().1.send_close_notify();
+                stream.flush().await?;
+                assert_eq!(stream.read(&mut [0]).await?, 0, "data after TLS close");
+                Ok::<_, io::Error>(())
+            })
+            .await;
+            assert!(
+                matches!(result, Ok(Ok(()))),
+                "{result:?}\n{}",
+                pair_logs(&client, &server)
+            );
+        } else {
+            exchange(client_port, target, &data)
+                .await
+                .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)));
+        }
+    }
+    let shoes = if peer_client { &server } else { &client };
+    if encrypted {
+        shoes.assert_handshakes(&["fresh", "resumed"]);
+    }
+    if vision {
+        for direction in ["READ", "WRITE"] {
+            assert_eq!(
+                shoes
+                    .logs()
+                    .matches(&format!(
+                        "VISION {direction}: Switching to direct copy mode"
+                    ))
+                    .count(),
+                2,
+                "{}",
+                shoes.logs()
+            );
+        }
+    }
+    // Force UDP frames across HTTP POST and native-record boundaries.
+    let udp = start_udp_echo_server_with_suffix("0.0.0.0", 0, b"reply")
+        .await
+        .unwrap();
+    let destination = (std::net::Ipv4Addr::LOCALHOST, udp.local_addr().port()).into();
+    for _ in 0..2 {
+        let association = timeout(
+            LIMIT,
+            Socks5UdpAssociation::connect("127.0.0.1", client_port),
+        )
+        .await
+        .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)))
+        .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)));
+        let payload = vec![37; 4096];
+        let reply = association
+            .send_to(destination, &payload)
+            .await
+            .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)));
+        assert_eq!(reply, [payload, b"reply".to_vec()].concat());
+    }
+    drop((client, server));
+    cover_task.stop().await;
+    target_task.stop().await;
+}
+
+#[tokio::test]
+async fn encrypted_xhttp_xray_client_resumes_and_carries_tcp_udp_and_vision() {
+    xhttp_matrix(Kind::Xray, true).await;
+}
+
+#[tokio::test]
+async fn encrypted_xhttp_xray_server_resumes_and_carries_tcp_udp_and_vision() {
+    xhttp_matrix(Kind::Xray, false).await;
+}
+
+#[tokio::test]
+async fn encrypted_xhttp_mihomo_client_resumes_and_carries_tcp_udp_and_vision() {
+    xhttp_matrix(Kind::Mihomo, true).await;
+}
+
+#[tokio::test]
+async fn encrypted_xhttp_mihomo_server_resumes_and_carries_tcp_udp_and_vision() {
+    xhttp_matrix(Kind::Mihomo, false).await;
+}
+
+async fn xhttp_matrix(peer: Kind, peer_client: bool) {
+    for version in ["plain", "h1", "h2c", "h2", "h3", "reality"] {
+        // Xray's client selects HTTP/1.1 without TLS; its server accepts h2c.
+        if peer == Kind::Xray && peer_client && version == "h2c" {
+            continue;
+        }
+        if peer == Kind::Mihomo && !peer_client && version == "h3" {
+            continue;
+        }
+        for mode in ["packet-up", "stream-up", "stream-one", "auto"] {
+            if mode == "auto" && !matches!(version, "h2" | "reality") {
+                continue;
+            }
+            for vision in [false, true] {
+                wrapped_case(
+                    peer,
+                    peer_client,
+                    Transport::XhttpMode { version, mode },
+                    true,
+                    vision,
+                )
+                .await;
+            }
+        }
+    }
+}
+
+#[tokio::test]
+async fn encrypted_non_vision_and_wss_resume_through_wrappers() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for client in [false, true] {
+            for transport in [
+                Transport::Tls,
+                Transport::Websocket,
+                Transport::Wss,
+                Transport::Reality,
+            ] {
+                wrapped_case(peer, client, transport, true, false).await;
+            }
+            wrapped_case(peer, client, Transport::Wss, true, true).await;
+        }
+    }
+    for vision in [false, true] {
+        wrapped_case(Kind::Shoes, false, Transport::Quic, true, vision).await;
+    }
+}
+
+#[tokio::test]
+async fn legacy_vless_websocket_and_mihomo_transports_carry_tcp_and_udp() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for client in [false, true] {
+            for transport in [Transport::Websocket, Transport::Wss] {
+                wrapped_case(peer, client, transport, false, false).await;
+            }
+        }
+    }
+    for client in [false, true] {
+        for transport in [Transport::Raw, Transport::Tls, Transport::Reality] {
+            wrapped_case(Kind::Mihomo, client, transport, false, false).await;
+        }
+    }
+}

tests/cases/optional_peer/vless_encryption/legacy.rs

@@ -1,5 +1,244 @@
 use super::*;
 
+pub(super) fn wrap_singbox(
+    mut config: Value,
+    server: bool,
+    transport: Transport,
+    cert: &str,
+    key: &str,
+    cover: u16,
+) -> Value {
+    let tls = if matches!(transport, Transport::Reality) {
+        let (private, public) = reality_keys();
+        if server {
+            json!({"enabled":true,"server_name":"localhost","reality":{"enabled":true,
+                "handshake":{"server":"localhost","server_port":cover},"private_key":private,"short_id":["0123456789abcdef"]}})
+        } else {
+            json!({"enabled":true,"server_name":"localhost","utls":{"enabled":true,"fingerprint":"chrome"},
+                "reality":{"enabled":true,"public_key":public,"short_id":"0123456789abcdef"}})
+        }
+    } else {
+        assert!(matches!(transport, Transport::Tls));
+        if server {
+            json!({"enabled":true,"certificate":cert.lines().collect::<Vec<_>>(),"key":key.lines().collect::<Vec<_>>()})
+        } else {
+            json!({"enabled":true,"server_name":"localhost","insecure":true})
+        }
+    };
+    config[if server { "inbounds" } else { "outbounds" }][0]["tls"] = tls;
+    config
+}
+
+fn enable_legacy_vision(config: &mut Value, server: bool, transport: Transport) {
+    let outer = shoes_protocol_mut(config, server);
+    let target = if !server {
+        outer
+    } else if matches!(transport, Transport::Tls) {
+        &mut outer["default_target"]
+    } else {
+        &mut outer["reality_targets"]["localhost"]
+    };
+    target["vision"] = json!(true);
+}
+
+async fn checked_transfer<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin>(
+    mut stream: S,
+    half_close_before_reply: bool,
+) -> io::Result<S> {
+    for _ in 0..2 {
+        stream.write_all(&[42; 64]).await?;
+        stream.flush().await?;
+        let mut reply = [0; 64];
+        stream.read_exact(&mut reply).await?;
+        assert_eq!(reply, [42; 64]);
+    }
+    let data: Vec<_> = (0..1024 * 1024 + 17).map(|i| (i % 251) as u8).collect();
+    let (mut r, mut w) = tokio::io::split(&mut stream);
+    let ((), output) = tokio::try_join!(
+        async {
+            w.write_all(&data).await?;
+            if half_close_before_reply {
+                w.shutdown().await?;
+            }
+            Ok::<_, io::Error>(())
+        },
+        async {
+            let mut output = Vec::new();
+            let mut buf = [0; 8192];
+            let mut next_pause = 65536;
+            while output.len() < data.len() {
+                let n = r.read(&mut buf).await?;
+                if n == 0 {
+                    break;
+                }
+                output.extend_from_slice(&buf[..n]);
+                assert!(output.len() <= data.len(), "duplicate data before EOF");
+                if output.len() >= next_pause {
+                    next_pause += 65536;
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+            }
+            if half_close_before_reply {
+                assert_eq!(r.read(&mut buf).await?, 0, "data after expected reply");
+            }
+            Ok::<_, io::Error>(output)
+        }
+    )?;
+    if output != data {
+        return Err(io::Error::other(format!(
+            "echo mismatch before EOF: received {} of {} bytes, first mismatch {:?}",
+            output.len(),
+            data.len(),
+            output.iter().zip(&data).position(|(a, b)| a != b),
+        )));
+    }
+    drop((r, w));
+    Ok(stream)
+}
+
+#[tokio::test]
+async fn legacy_peers_assert_direct_end_and_eof_over_tls_and_reality() {
+    for peer in [Kind::Xray, Kind::Mihomo, Kind::SingBox] {
+        for peer_client in [false, true] {
+            for transport in [Transport::Tls, Transport::Reality] {
+                for inner_tls12 in [None, Some(true), Some(false)] {
+                    // sing-box's Vision wrapper has no CloseWrite. Clients close both sides on
+                    // upload EOF; servers can reset TLS 1.2 replies when the target closes first.
+                    // Upstream fix: https://github.com/SagerNet/sing-vmess/pull/16
+                    let half_close_before_reply =
+                        peer != Kind::SingBox || (!peer_client && inner_tls12 != Some(true));
+                    decision_case(
+                        peer,
+                        peer_client,
+                        transport,
+                        inner_tls12,
+                        half_close_before_reply,
+                    )
+                    .await;
+                }
+            }
+        }
+    }
+}
+
+#[tokio::test]
+#[ignore = "pinned sing-box Vision client closes both directions on upload EOF"]
+async fn singbox_vision_client_half_close_preserves_pending_reply() {
+    decision_case(Kind::SingBox, true, Transport::Tls, None, true).await;
+}
+
+#[tokio::test]
+#[ignore = "pinned sing-box Vision server can reset TLS 1.2 replies during upload half-close"]
+async fn singbox_vision_server_half_close_preserves_tls12_reply() {
+    for transport in [Transport::Tls, Transport::Reality] {
+        decision_case(Kind::SingBox, false, transport, Some(true), true).await;
+    }
+}
+
+async fn decision_case(
+    peer: Kind,
+    peer_client: bool,
+    transport: Transport,
+    inner_tls12: Option<bool>,
+    half_close_before_reply: bool,
+) {
+    eprintln!(
+        "legacy decisions: {peer:?} client={peer_client} {transport:?} inner_tls12={inner_tls12:?}"
+    );
+    let (server_kind, client_kind) = peer_roles(peer, peer_client);
+    let (cover, _, mut cover_task) = tls_echo(false).await;
+    let (target, connector, mut target_task) = if let Some(tls12) = inner_tls12 {
+        let (port, connector, task) = tls_echo(tls12).await;
+        (port, Some(connector), task)
+    } else {
+        let (port, _, task) = echo().await;
+        (port, None, task)
+    };
+    let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
+    let (cert, key) = (cert.cert.pem(), cert.signing_key.serialize_pem());
+    let server_port = free_port().await;
+    let client_port = free_port().await;
+    let mut server_settings = wrap(
+        server_kind,
+        server_config(server_kind, server_port, "none", server_kind != Kind::Shoes),
+        true,
+        transport,
+        &cert,
+        &key,
+        cover,
+    );
+    let mut client_settings = wrap(
+        client_kind,
+        client_config(
+            client_kind,
+            client_port,
+            server_port,
+            "none",
+            client_kind != Kind::Shoes,
+        ),
+        false,
+        transport,
+        &cert,
+        &key,
+        cover,
+    );
+    if server_kind == Kind::Shoes {
+        enable_legacy_vision(&mut server_settings, true, transport);
+    } else {
+        enable_legacy_vision(&mut client_settings, false, transport);
+    }
+    let server = start(server_kind, server_port, server_settings).await;
+    let client = start(client_kind, client_port, client_settings).await;
+    timeout(Duration::from_secs(20), async {
+        let socket = socks_connect(client_port, target).await?;
+        if let Some(connector) = connector {
+            let mut tls = checked_transfer(
+                connector
+                    .connect("localhost".try_into().unwrap(), socket)
+                    .await?,
+                half_close_before_reply,
+            )
+            .await?;
+            if !half_close_before_reply {
+                tls.get_mut().1.send_close_notify();
+                tls.flush().await?;
+                assert_eq!(tls.read(&mut [0]).await?, 0, "data after TLS close");
+            }
+        } else {
+            let mut socket = checked_transfer(socket, half_close_before_reply).await?;
+            if !half_close_before_reply {
+                socket.shutdown().await?;
+                assert_eq!(socket.read(&mut [0]).await?, 0, "data after expected reply");
+            }
+        }
+        Ok::<_, io::Error>(())
+    })
+    .await
+    .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)))
+    .unwrap_or_else(|e| panic!("{e}\n{}", pair_logs(&client, &server)));
+    let shoes = if peer_client { &server } else { &client };
+    let direct = inner_tls12 == Some(false);
+    for direction in ["READ", "WRITE"] {
+        let logs = shoes.logs();
+        assert_eq!(
+            logs.contains(&format!(
+                "VISION {direction}: Switching to direct copy mode"
+            )),
+            direct,
+            "{logs}"
+        );
+        if !direct {
+            assert!(
+                logs.contains(&format!("VISION {direction}: Switching to Tls mode")),
+                "{logs}"
+            );
+        }
+    }
+    drop((client, server));
+    cover_task.stop().await;
+    target_task.stop().await;
+}
+
 #[derive(Debug, PartialEq, Eq)]
 struct LegacyObservation {
     eof: bool,

tests/cases/optional_peer/vless_encryption/lifecycle.rs

@@ -0,0 +1,216 @@
+use super::*;
+
+#[cfg(unix)]
+async fn reload(peer: &Peer, config: &Value) {
+    let ready = peer.logs().matches("Servers ready").count();
+    let path = peer.directory.path().join("config.json");
+    std::fs::write(path, serde_json::to_vec(config).unwrap()).unwrap();
+    assert_eq!(
+        unsafe { libc::kill(peer.process.id() as i32, libc::SIGHUP) },
+        0
+    );
+    timeout(LIMIT, async {
+        while peer.logs().matches("Servers ready").count() == ready {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|e| panic!("reload: {e}\n{}", peer.logs()));
+}
+
+async fn roundtrip(stream: &mut TcpStream, payload: &[u8]) {
+    timeout(LIMIT, async {
+        stream.write_all(payload).await.unwrap();
+        let mut reply = vec![0; payload.len()];
+        stream.read_exact(&mut reply).await.unwrap();
+        assert_eq!(reply, payload);
+    })
+    .await
+    .unwrap();
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn live_reload_resets_tickets_preserves_active_streams_and_rotates_keys() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for mode in ["native", "xorpub", "random"] {
+            let (encryption, decryption, _) = credentials(mode, &["x25519"]);
+            let (target, count, mut target_task) = echo().await;
+            let port = free_port().await;
+            let mut config = server_config(Kind::Shoes, port, &decryption, false);
+            let server = start(Kind::Shoes, port, config.clone()).await;
+            let client_port = free_port().await;
+            let client = start(
+                peer,
+                client_port,
+                client_config(peer, client_port, port, &encryption, false),
+            )
+            .await;
+            let mut active = socks_connect(client_port, target).await.unwrap();
+            roundtrip(&mut active, b"old generation").await;
+            exchange(client_port, target, b"warm").await.unwrap();
+            server.assert_handshakes(&["fresh", "resumed"]);
+
+            config[0]["protocol"]["decryption"] = json!(decryption.replace(".600s.", ".0s."));
+            reload(&server, &config).await;
+            roundtrip(&mut active, b"still live after reload").await;
+            assert!(
+                exchange(client_port, target, b"stale must not arrive")
+                    .await
+                    .is_err()
+            );
+            assert_eq!(count.load(Ordering::SeqCst), 2);
+            for _ in 0..2 {
+                exchange(client_port, target, b"fresh-only recovery")
+                    .await
+                    .unwrap();
+            }
+            server.assert_handshakes(&["fresh", "resumed", "fresh", "fresh"]);
+
+            let (next_encryption, next_decryption, _) = credentials(mode, &["x25519"]);
+            config[0]["protocol"]["decryption"] = json!(next_decryption);
+            reload(&server, &config).await;
+            roundtrip(&mut active, b"old stream survives key rotation").await;
+            assert!(
+                exchange(client_port, target, b"old key must fail")
+                    .await
+                    .is_err()
+            );
+            assert_eq!(count.load(Ordering::SeqCst), 4);
+            drop(client);
+            let client = start(
+                peer,
+                client_port,
+                client_config(peer, client_port, port, &next_encryption, false),
+            )
+            .await;
+            for _ in 0..2 {
+                exchange(client_port, target, b"new key").await.unwrap();
+            }
+            server.assert_handshakes(&["fresh", "resumed", "fresh", "fresh", "fresh", "resumed"]);
+            drop((active, client, server));
+            target_task.stop().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn expired_tickets_use_a_fresh_handshake_without_losing_the_request() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for peer_client in [false, true] {
+            let (server_kind, client_kind) = peer_roles(peer, peer_client);
+            let (encryption, decryption, _) = credentials("random", &["x25519"]);
+            let decryption = decryption.replace(".600s.", ".5-5s.");
+            let (target, count, mut task) = echo().await;
+            let port = free_port().await;
+            let server = start(
+                server_kind,
+                port,
+                server_config(server_kind, port, &decryption, false),
+            )
+            .await;
+            let proxy = free_port().await;
+            let client = start(
+                client_kind,
+                proxy,
+                client_config(client_kind, proxy, port, &encryption, false),
+            )
+            .await;
+            for _ in 0..2 {
+                exchange(proxy, target, b"warm before expiry")
+                    .await
+                    .unwrap();
+            }
+            let shoes = if peer_client { &server } else { &client };
+            shoes.assert_handshakes(&["fresh", "resumed"]);
+            tokio::time::sleep(Duration::from_millis(5200)).await;
+            exchange(proxy, target, b"fresh after expiry")
+                .await
+                .unwrap();
+            shoes.assert_handshakes(&["fresh", "resumed", "fresh"]);
+            assert_eq!(count.load(Ordering::SeqCst), 3);
+            drop((client, server));
+            task.stop().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn shoes_client_rejects_stale_peer_ticket_without_replaying_application_data() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for mode in ["native", "xorpub", "random"] {
+            let (encryption, decryption, _) = credentials(mode, &["x25519"]);
+            let (target, count, mut task) = echo().await;
+            let port = free_port().await;
+            let config = server_config(peer, port, &decryption, false);
+            let server = start(peer, port, config.clone()).await;
+            let proxy = free_port().await;
+            let client = start(
+                Kind::Shoes,
+                proxy,
+                client_config(Kind::Shoes, proxy, port, &encryption, false),
+            )
+            .await;
+            for _ in 0..2 {
+                exchange(proxy, target, b"warm").await.unwrap();
+            }
+            client.assert_handshakes(&["fresh", "resumed"]);
+            drop(server);
+            let server = start(peer, port, config).await;
+            assert!(
+                exchange(proxy, target, b"do not retry this request")
+                    .await
+                    .is_err()
+            );
+            assert_eq!(count.load(Ordering::SeqCst), 2);
+            exchange(proxy, target, b"independent recovery")
+                .await
+                .unwrap();
+            assert_eq!(count.load(Ordering::SeqCst), 3);
+            client.assert_handshakes(&["fresh", "resumed", "resumed", "fresh"]);
+            drop((client, server));
+            task.stop().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn live_chacha_peer_clients_complete_fresh_and_resumed_handshakes() {
+    for peer in [Kind::Xray, Kind::Mihomo] {
+        for mode in ["native", "random"] {
+            let (encryption, decryption, _) = credentials(mode, &["mlkem768"]);
+            let (target, _, mut task) = echo().await;
+            let port = free_port().await;
+            let server = start(
+                Kind::Shoes,
+                port,
+                server_config(Kind::Shoes, port, &decryption, false),
+            )
+            .await;
+            let proxy = free_port().await;
+            let client = start_options(
+                peer,
+                proxy,
+                client_config(peer, proxy, port, &encryption, false),
+                peer.binary(),
+                Some("cpu.aes=off,cpu.pclmulqdq=off"),
+            )
+            .await;
+            for _ in 0..2 {
+                exchange(proxy, target, &vec![42; 32768]).await.unwrap();
+            }
+            server.assert_handshakes(&["fresh", "resumed"]);
+            assert_eq!(
+                server
+                    .logs()
+                    .matches("VLESS encryption server cipher: ChaCha20Poly1305")
+                    .count(),
+                2,
+                "{}",
+                server.logs()
+            );
+            drop((client, server));
+            task.stop().await;
+        }
+    }
+}

tests/cases/optional_peer/wireguard_interop.rs

@@ -29,7 +29,7 @@ fn singbox_server(port: u16) -> Value {
 }
 
 #[tokio::test]
-#[ignore = "Requires Xray 26.10.10; set SHOES_TEST_XRAY_BIN"]
+#[ignore = "Requires Xray 26.10.17; set SHOES_TEST_XRAY_BIN"]
 async fn wireguard_xray_both_roles_tcp_udp_ipv4_ipv6() -> io::Result<()> {
     tokio::time::timeout(Duration::from_secs(90), async {
         for shoes_is_server in [false, true] {

tests/fixtures/vless_encryption/README.md

@@ -37,15 +37,61 @@ GOMAXPROCS=1 go run -p 1 /path/to/shoes/tests/fixtures/vless_encryption/generate
   --verify-rust "$HOME/vless-rust-vectors.json"
 ```
 
-The native peer suite uses `XRAY_BINARY`, `MIHOMO_BINARY` and optionally
-`SHOES_TEST_SHOES_BIN`. Missing requested peers fail rather than skip. Its wire
-tap distinguishes fresh from resumed handshakes and checks raw DIRECT lengths.
+The peer suite uses `XRAY_BINARY`, `MIHOMO_BINARY`, `SHOES_TEST_SING_BOX_BIN`, and
+optionally `SHOES_TEST_SHOES_BIN`. CI pins Xray 26.10.17
+(`92ccc939aaf4f8965f369cf0b142e005aeaa11d5`), including its Vision TLS-buffer
+compatibility fix. The committed vectors retain their original provenance above.
+Missing requested peers fail rather than skip. The wire tap distinguishes fresh
+from resumed handshakes and checks raw DIRECT lengths. Successful transfers plus
+secret-free debug diagnostics verify resumption inside opaque carriers.
 Xray's command-2 replies are limited to 8190 bytes; its XUDP writer limits
 payloads to 7526 bytes. Peer UDP tests use that intersection. Separate Shoes
 tests cover larger XUDP frames and preservation of the existing empty-DATA
 and control-frame semantics. Sing-box has no native encryption support and is
 used only by the legacy TLS/REALITY regression suites.
 
+The enabled matrix covers:
+
+- Native TCP and UDP over TLS, WebSocket, WSS, REALITY, and Shoes raw QUIC.
+- Encrypted XHTTP with and without Vision: packet-up, stream-up, and stream-one
+  over H1, H2C, H2, H3, and REALITY; auto mode over H2 and REALITY. Xray cannot
+  initiate H2C, and Mihomo cannot serve H3; those directions are excluded.
+- Legacy plain VLESS over WS/WSS and Mihomo raw/TLS/REALITY. Legacy Vision uses
+  Xray, Mihomo, and sing-box in both roles, asserting DIRECT for inner TLS 1.3,
+  END for plain/TLS 1.2 traffic, exact payloads under slow reads, and remote EOF.
+  Clients half-close before receiving the full reply except in the sing-box cases
+  below.
+  The pinned sing-box Vision client closes both directions on upload EOF, so its
+  enabled cases finish receiving before closing. The ignored
+  `singbox_vision_client_half_close_preserves_pending_reply` retains that failing
+  peer reproducer. The sing-box Vision server can also reset inner TLS 1.2 replies
+  when the target closes before upload EOF arrives. Only those server cases, over
+  TLS and REALITY, receive the full payload before sending inner `close_notify`
+  and requiring clean EOF. The ignored
+  `singbox_vision_server_half_close_preserves_tls12_reply` retains the intermittent
+  server reproducer. Remove this server workaround when the pinned peer includes
+  the Vision half-close fix in
+  [SagerNet/sing-vmess#16](https://github.com/SagerNet/sing-vmess/pull/16).
+  Plain and TLS 1.3 server cases still assert early half-close.
+  All native-Vision peer cases close inner TLS gracefully without
+  requiring transport-level half-close support: WebSocket and Mihomo's masked
+  native wrapper can close both transport halves. Native TCP half-close and a
+  delayed reverse reply are asserted separately in the library/splice tests.
+- Live server reload, zero ticket lifetime, key rotation, expiry, peer restart,
+  and recovery on a new request without replaying failed application data.
+- Live ChaCha20-Poly1305 from peer clients with Go AES acceleration disabled.
+- Actual TLS/SOCKS and TLS/AnyTLS consumers chained over native encryption,
+  including slow full-duplex reads and cancellation of another stream.
+
+The library suite complements the process matrix with native fresh/resumed
+Vision decisions, fragmented and malformed records, the maximum native record
+across Vision's 8 KiB buffer, real Linux splice transfers and masked/wrapped
+negative controls, concurrent duplicate flights across bind IPs, SNI/WS ticket
+isolation, IPv4/IPv6/domain destinations, empty UDP replies, and native-Vision
+H2MUX rejection. IPv6 address cases run when IPv6 loopback is available;
+IPv4 and domain cases always run without relying on dual-stack sockets.
+These tests also run without optional peer binaries.
+
 The ignored `legacy_baseline_matches_backpressure_and_half_close` test compares
 both Shoes roles against Xray over TLS, including slow reads and half-close
 after DIRECT. It verifies payload integrity, observes remote EOF within five

tests/support/src/xray.rs

@@ -11,7 +11,7 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 use crate::process::ProcessGuard;
 
-pub const VERSION: &str = "26.10.10";
+pub const VERSION: &str = "26.10.17";
 
 pub struct Peer {
     _process: ProcessGuard,