alksocks 0.3.2

//! The cross-cutting assembly integration suite: the whole assembly
//! layer's behavior pinned in one place (the batch-2 closing task; the
//! architecture-sync check's test half).
//!
//! The per-component tasks own their verticals (tests/door-*.rs,
//! tests/local-*.rs, tests/composed-*.rs); this suite owns the
//! compositions and the cross-shape invariants:
//!
//! - **The full `-D` story end to end**: a vanilla tun2proxy-shaped
//!   client → the cred-bearing TCP door → the channels-native session
//!   → the producer with `local` providers → real echo targets:
//!   CONNECT + UDP (the RFC-header'd datagram through the UDP door) +
//!   BIND (the vanilla BIND conversation through the door) all through
//!   one door process against one producer registration (the
//!   one-registration-three-commands regression, now through the
//!   doors).
//! - **The composed story end to end**: the producer with
//!   `composed::direct_egress` + `composed::bindfirst_accept`:
//!   CONNECT through the direct op, ASSOCIATE through the lazy table,
//!   BIND through the bind-first listener (the ADR-008 composed
//!   variant set, all three commands).
//! - **Mixed assembly**: one producer registration serving
//!   channels-native sessions AND door clients AND composed seams
//!   concurrently (the producer/consumer-roles-not-sides posture:
//!   both shapes on one connection).
//! - **The teardown matrix, assembly edition**: door-client disconnect
//!   tears the association/channel down (no leaks: the channel count
//!   witnesses); the idle bounds fire; fresh sessions succeed
//!   immediately after each teardown (no connection poisoning).
#![cfg(all(feature = "local", feature = "composed"))]

mod harness;

fn init_trace() {
    let _ = tracing_subscriber::fmt()
        .with_max_level(tracing::level_filters::LevelFilter::DEBUG)
        .with_test_writer()
        .try_init();
}

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use alkcall::channels::operations::ChannelCore;
use alkcall::core::auth::{Identity, IdentityProvider};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream, UdpSocket};
use tokio::time::timeout;

use alksocks::composed::{bindfirst_accept, direct_egress, register_forwarded_serving};
use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession};
use alksocks::rfc1928::consts;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use alksocks::{CredentialTable, DoorCredential, FrontDoor, UdpFrontDoor, SOCKS5_BIND_SCOPE};

use harness::{
    composed_full_opener_identity, composed_push_accept, wire_bindfirst_composed,
    wire_consumer_side_with, wire_consumer_side_with_provider, COMPOSED_BOUND_PORT,
    COMPOSED_PEER_PORT, COMPOSED_RESOURCE,
};

const BOUND: Duration = Duration::from_secs(10);

// --- Real targets ------------------------------------------------------------

/// A real TCP echo server on an ephemeral loopback port.
async fn echo_tcp_server() -> SocketAddr {
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let addr = listener.local_addr().expect("local_addr");
    tokio::spawn(async move {
        loop {
            let Ok((mut stream, _)) = listener.accept().await else {
                return;
            };
            tokio::spawn(async move {
                let mut buf = [0u8; 4096];
                loop {
                    match stream.read(&mut buf).await {
                        Ok(0) | Err(_) => return,
                        Ok(n) => {
                            if stream.write_all(&buf[..n]).await.is_err() {
                                return;
                            }
                        }
                    }
                }
            });
        }
    });
    addr
}

/// A real UDP echo server on an ephemeral loopback port.
async fn echo_udp_server() -> SocketAddr {
    let sock = UdpSocket::bind(("127.0.0.1", 0))
        .await
        .expect("bind udp echo");
    let addr = sock.local_addr().expect("addr");
    tokio::spawn(async move {
        let mut buf = vec![0u8; 65535];
        loop {
            let Ok((n, peer)) = sock.recv_from(&mut buf).await else {
                return;
            };
            if sock.send_to(&buf[..n], peer).await.is_err() {
                return;
            }
        }
    });
    addr
}

// --- Vanilla-client helpers (the door boundary's RFC forms) ------------------

/// The RFC 1928 cred-bearing conversation's CONNECT: greeting (the
/// password method offered) → method selection → credentials → the
/// CONNECT command for a v4 target → the reply → (code, stream, BND).
/// The vanilla tun2proxy shape (`--proxy socks5://user:pass@host:port`).
async fn vanilla_cred_connect(door: SocketAddr, target: SocketAddr) -> (TcpStream, u8, [u8; 6]) {
    let mut stream = vanilla_with_creds(door).await;
    let mut req = vec![0x05, 0x01, 0x00, 0x01];
    let SocketAddr::V4(v4) = target else {
        panic!("v4 targets only")
    };
    req.extend_from_slice(&v4.ip().octets());
    req.extend_from_slice(&v4.port().to_be_bytes());
    stream.write_all(&req).await.expect("request");
    stream.flush().await.expect("flush");

    let mut head = [0u8; 4];
    stream.read_exact(&mut head).await.expect("reply head");
    assert_eq!(head[0], 0x05, "reply is v5");
    assert_eq!(head[3], 0x01, "v4 BND.ADDR");
    let mut addr = [0u8; 6];
    stream.read_exact(&mut addr).await.expect("BND.ADDR");
    (stream, head[1], addr)
}

/// The RFC 1928 NoAuth conversation against a TCP door: greet, send a
/// CONNECT command for a v4 target, read the reply → (code, stream).
async fn vanilla_no_auth_connect(door: SocketAddr, target: SocketAddr) -> (TcpStream, u8, [u8; 6]) {
    let mut stream = TcpStream::connect(door).await.expect("door connect");
    stream
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    stream
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    assert_eq!(chosen, [0x05, 0x00], "no-auth selected");

    let mut req = vec![0x05, 0x01, 0x00, 0x01];
    let SocketAddr::V4(v4) = target else {
        panic!("v4 targets only")
    };
    req.extend_from_slice(&v4.ip().octets());
    req.extend_from_slice(&v4.port().to_be_bytes());
    stream.write_all(&req).await.expect("request");
    stream.flush().await.expect("flush");

    let mut head = [0u8; 4];
    stream.read_exact(&mut head).await.expect("reply head");
    assert_eq!(head[0], 0x05, "reply is v5");
    assert_eq!(head[3], 0x01, "v4 BND.ADDR");
    let mut addr = [0u8; 6];
    stream.read_exact(&mut addr).await.expect("BND.ADDR");
    (stream, head[1], addr)
}

/// An RFC-header'd datagram (the vanilla client's sendto form, the
/// door boundary's codec, hand-rolled per the RFC wire form).
fn rfc_datagram(target: SocketAddr, payload: &[u8]) -> Vec<u8> {
    let SocketAddr::V4(v4) = target else {
        panic!("v4 targets only")
    };
    let mut dg = vec![0u8, 0, 0, 0x01];
    dg.extend_from_slice(&v4.ip().octets());
    dg.extend_from_slice(&v4.port().to_be_bytes());
    dg.extend_from_slice(payload);
    dg
}

/// Parse an RFC-header'd datagram back → (addr, payload).
fn parse_rfc_datagram(wire: &[u8]) -> (SocketAddr, &[u8]) {
    assert_eq!(&wire[..2], &[0, 0], "RSV zero");
    assert_eq!(wire[3], 0x01, "v4 ATYP");
    let ip = std::net::Ipv4Addr::new(wire[4], wire[5], wire[6], wire[7]);
    let port = u16::from_be_bytes([wire[8], wire[9]]);
    (SocketAddr::from((ip, port)), &wire[10..])
}

// --- The token provider (the cred-bearing door's CF-005 seam) ----------------

/// Resolves the door's mapped identity ids (`door-`-prefixed) to
/// identities carrying the requested scopes; anything else resolves to
/// none (the CF-005 precedence fall-through).
#[derive(Clone)]
struct DoorTokenProvider {
    bind_scoped: Vec<String>,
}

impl IdentityProvider for DoorTokenProvider {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
        let id = String::from_utf8_lossy(&token.raw).into_owned();
        if id.starts_with("door-") {
            let mut scopes = vec![alksocks::SOCKS5_OPEN_SCOPE.to_string()];
            if self.bind_scoped.contains(&id) {
                scopes.push(SOCKS5_BIND_SCOPE.to_string());
            }
            return Some(Identity {
                id,
                scopes,
                resources: Default::default(),
            });
        }
        None
    }
}

fn cred_table() -> CredentialTable {
    CredentialTable::new(vec![DoorCredential {
        username: "vanilla".to_string(),
        password: "hunter2".to_string(),
        identity: Identity {
            id: "door-vanilla".to_string(),
            scopes: vec![
                alksocks::SOCKS5_OPEN_SCOPE.to_string(),
                SOCKS5_BIND_SCOPE.to_string(),
            ],
            resources: Default::default(),
        },
    }])
}

/// The RFC 1929 cred-bearing conversation: greeting (password method
/// offered) → method selection → credentials → command. Returns the
/// post-auth stream.
async fn vanilla_with_creds(door: SocketAddr) -> TcpStream {
    let mut stream = TcpStream::connect(door).await.expect("door connect");
    stream
        .write_all(&[0x05, 0x01, 0x02])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    stream
        .read_exact(&mut chosen)
        .await
        .expect("method selected");
    assert_eq!(chosen, [0x05, 0x02], "password auth selected");
    let mut packet = vec![0x01, 7];
    packet.extend_from_slice(b"vanilla");
    packet.push(7);
    packet.extend_from_slice(b"hunter2");
    stream.write_all(&packet).await.expect("creds sent");
    let mut status = [0u8; 2];
    stream.read_exact(&mut status).await.expect("status");
    assert_eq!(status, [0x01, 0x00], "RFC 1929 accepted");
    stream
}

// --- The composed suite (the direct-op + bind-first producer) ----------------

/// The composed topology: the SOCKS5 producer serves the direct op
/// (CONNECT + ASSOCIATE's egress) AND the bind-first op (BIND's
/// listener) on one connection: the ADR-008 composed variant set.
/// The direct-op dial is a real TCP/UDP-capable stand-in: a TCP dial
/// dials the real echo target (CONNECT rides it); a UDP dial yields a
/// framed echo pipe (ASSOCIATE's destinations echo datagrams).
struct ComposedSuite {
    topo: harness::BindFirstTopology,
    queue: harness::ComposedAcceptQueue,
    dial_count: Arc<std::sync::atomic::AtomicUsize>,
}

async fn wire_composed_suite() -> ComposedSuite {
    init_trace();
    let resource_registry = alktunnels::producer::ResourceRegistry::new();
    resource_registry
        .register(
            COMPOSED_RESOURCE,
            alktunnels::params::Substrate::Tcp,
            "assembly-owned",
        )
        .await;
    let queue = harness::composed_fresh_queue();
    let dial_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));

    // The direct-op dial: a TCP dial rides the real sockets via
    // `local::tcp_dial`-shaped behavior re-expressed on the alktunnels
    // `DialFn` (the direct op's TargetHandle); a UDP dial yields the
    // framed echo pipe (the composed-egress suite's shape).
    let dial = {
        let dial_count = Arc::clone(&dial_count);
        Arc::new(
            move |substrate: alktunnels::params::Substrate,
                  backing: &str|
                  -> futures::future::BoxFuture<
                'static,
                Result<alktunnels::producer::TargetHandle, alktunnels::TunnelEstablishError>,
            > {
                let dial_count = Arc::clone(&dial_count);
                let backing = backing.to_string();
                Box::pin(async move {
                    use alktunnels::params::Substrate as S;
                    use alktunnels::producer::TargetHandle as TunnelTargetHandle;
                    use alktunnels::TunnelEstablishError;
                    dial_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
                    match substrate {
                        S::Tcp => {
                            // A real TCP dial: `host:port` → TcpStream → split.
                            let (host, port) = backing.rsplit_once(':').ok_or_else(|| {
                                TunnelEstablishError::DialFailed("bad target".into())
                            })?;
                            let addr: SocketAddr =
                                format!("{host}:{port}").parse().map_err(|e| {
                                    TunnelEstablishError::DialFailed(format!("parse: {e}"))
                                })?;
                            let stream = TcpStream::connect(addr).await.map_err(|e| {
                                TunnelEstablishError::DialFailed(format!("connect: {e}"))
                            })?;
                            let (r, w) = stream.into_split();
                            Ok(TunnelTargetHandle {
                                read: Box::new(r),
                                write: Box::new(w),
                            })
                        }
                        S::Udp => {
                            // The framed UDP echo pipe (the composed-egress
                            // suite's shape): the "target" decodes the
                            // tunnel codec and echoes each datagram.
                            let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
                            tokio::spawn(async move {
                                use tokio::io::AsyncReadExt;
                                let (mut t_read, mut t_write) = tokio::io::split(target_side);
                                let mut reader = alktunnels::wire::DatagramReader::new();
                                let mut buf = vec![0u8; 16 * 1024];
                                loop {
                                    let n = match t_read.read(&mut buf).await {
                                        Ok(0) | Err(_) => return,
                                        Ok(n) => n,
                                    };
                                    for dg in reader.feed(&buf[..n]) {
                                        let framed = match alktunnels::wire::frame_datagram(&dg) {
                                            Ok(f) => f,
                                            Err(_) => return,
                                        };
                                        if t_write.write_all(&framed).await.is_err() {
                                            return;
                                        }
                                        let _ = t_write.flush().await;
                                    }
                                }
                            });
                            let (c_read, c_write) = tokio::io::split(consumer_side);
                            Ok(TunnelTargetHandle {
                                read: Box::new(c_read),
                                write: Box::new(c_write),
                            })
                        }
                        other => Err(TunnelEstablishError::DialFailed(format!(
                            "substrate not wired: {other:?}"
                        ))),
                    }
                })
            },
        ) as alktunnels::producer::DialFn
    };

    let topo = wire_bindfirst_composed(
        resource_registry,
        harness::composed_test_bind(),
        harness::composed_queue_loop(queue.clone()),
        None, // the SOCKS5 openable registers post-hoc, below
        composed_full_opener_identity(),
    )
    .await;

    // The forwarded serving half rides the opener's registry.
    let registry = register_forwarded_serving(
        &ChannelCore::new(
            topo.opener.manager().clone(),
            alkcall::channels::policy::default_policy(),
        ),
        &topo.opener_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
    )
    .expect("register forwarded openable");

    // The SOCKS5 producer: the composed accept seam (BIND) + the
    // composed direct-op egress (ASSOCIATE) + a real TCP dial
    // (CONNECT: the direct-op dial's TCP arm shares the real target;
    // the CONNECT dial here is the crate's own local dial over the
    // SOCKS5 seam, distinct from the tunnel dial).
    let factory = bindfirst_accept(topo.opener.clone(), registry, COMPOSED_RESOURCE);
    let core = ChannelCore::new(
        topo.producer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    alksocks::producer::Socks5Producer::builder()
        .dial(alksocks::local::tcp_dial())
        .egress(direct_egress(topo.opener.clone(), 8))
        .accept(factory)
        .build()
        .register(
            &core,
            &topo.producer_registry,
            alkcall::core::auth::AuthContext::anonymous(b"alk/socks5"),
            None,
        )
        .expect("register socks5 openable (the composed suite's producer)");

    // The direct op rides the producer's registry (the direct-op
    // caller is the opener's client; the dial is the tunnel dial
    // above).
    let direct_core = ChannelCore::new(
        topo.producer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    alktunnels::producer::register_tunnel_direct_openable(
        &direct_core,
        &topo.producer_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
        dial,
        None,
    )
    .expect("register the tunnel direct op (the composed suite)");

    ComposedSuite {
        topo,
        queue,
        dial_count,
    }
}

// --- The suite -----------------------------------------------------------------

/// THE full `-D` story (the one-registration-three-commands regression
/// through the doors): one producer registration (the consumer-side
/// topology over the `local` providers) + one cred-bearing TCP door +
/// one UDP door. A vanilla client drives all three commands through
/// the doors against the one registration:
///
/// - CONNECT: the RFC 1929 door conversation → the CONNECT request →
///   the real TCP echo target round-trips.
/// - BIND: the door's BIND conversation (its own RFC session, the
///   vanilla client's one-command-per-connection shape) → the app
///   server dials the advertised address → reply#2's real peer → the
///   data plane.
/// - ASSOCIATE: the UDP door's relay socket ← the RFC-header'd
///   datagram → the real UDP echo target → back RFC-header'd.
///
/// All three sessions are live simultaneously mid-life of each other;
/// each teardown ends only its own session. The two doors ride two
/// channels connections to the same producer registration (the
/// assembly story: one registration, many connections. The cred-
/// bearing door's `TokenClient` takes its client's call connection,
/// the door-only contract, so the UDP door's association rides its
/// own).
#[tokio::test]
async fn full_dash_d_story_one_registration_three_commands_through_the_doors() {
    init_trace();
    let tcp_target = echo_tcp_server().await;
    let udp_target = echo_udp_server().await;

    // One producer registration, two door connections: the cred-
    // bearing TCP door rides one client (its TokenClient takes the
    // call connection); the UDP door's association rides a second
    // connection to the same registration (the UDP relay is a
    // separate datagram plane by construction: one relay socket = one
    // association).
    let tcp_dial = alksocks::local::tcp_dial();
    let tcp_topo = wire_consumer_side_with_provider(
        tcp_dial,
        Some(alksocks::local::udp_egress()),
        Some(alksocks::local::bind_accept([127, 0, 0, 1])),
        Some(harness::consumer_identity()),
        Arc::new(DoorTokenProvider {
            bind_scoped: vec!["door-vanilla".to_string()],
        }),
    )
    .await;
    let udp_dial = alksocks::local::tcp_dial();
    let udp_topo = wire_consumer_side_with_provider(
        udp_dial,
        Some(alksocks::local::udp_egress()),
        Some(alksocks::local::bind_accept([127, 0, 0, 1])),
        Some(harness::consumer_identity()),
        Arc::new(DoorTokenProvider {
            bind_scoped: vec!["door-vanilla".to_string()],
        }),
    )
    .await;

    // The cred-bearing TCP door (the CONNECT + BIND shape) + the UDP
    // door (the datagram plane): one door process per plane, the
    // assembly wiring consumer.md documents.
    let tcp_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let door_addr = tcp_listener.local_addr().expect("door addr");
    let door = FrontDoor::builder()
        .credentials(cred_table())
        .bind(door_addr)
        .build()
        .expect("the door builds");
    let _tcp_serve = tokio::spawn({
        let client = Arc::clone(&tcp_topo.consumer);
        async move {
            door.serve_on(tcp_listener, client)
                .await
                .expect("the door serves")
        }
    });
    let udp_relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let udp_relay_addr = udp_relay.local_addr().expect("relay addr");
    let udp_door = UdpFrontDoor::builder()
        .bind(udp_relay_addr)
        .build()
        .expect("the door builds");
    let _udp_serve = tokio::spawn({
        let client = Arc::clone(&udp_topo.consumer);
        async move {
            udp_door
                .serve_on(udp_relay, client)
                .await
                .expect("the udp door serves")
        }
    });

    // --- CONNECT through the cred-bearing door (live) ---------------
    let mut connect_stream = timeout(BOUND, vanilla_with_creds(door_addr))
        .await
        .expect("the cred conversation");
    let mut req = vec![0x05, 0x01, 0x00, 0x01];
    let SocketAddr::V4(v4) = tcp_target else {
        panic!("v4 target")
    };
    req.extend_from_slice(&v4.ip().octets());
    req.extend_from_slice(&v4.port().to_be_bytes());
    connect_stream
        .write_all(&req)
        .await
        .expect("connect request");
    let mut head = [0u8; 4];
    timeout(BOUND, connect_stream.read_exact(&mut head))
        .await
        .expect("reply#1")
        .expect("read");
    assert_eq!(
        head[1],
        consts::SOCKS5_REPLY_SUCCEEDED,
        "CONNECT through the door"
    );
    let mut connect_bnd = [0u8; 6];
    timeout(BOUND, connect_stream.read_exact(&mut connect_bnd))
        .await
        .expect("the CONNECT reply's BND.ADDR")
        .expect("read");

    // --- BIND through the door (live, mid-life of the CONNECT) ------
    // Its own door connection (the vanilla one-command-per-connection
    // shape: the door's RFC half reads one command per conversation).
    let mut bind_stream = timeout(BOUND, vanilla_with_creds(door_addr))
        .await
        .expect("the second cred conversation");
    bind_stream
        .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("bind request");
    let mut head1 = [0u8; 4];
    timeout(BOUND, bind_stream.read_exact(&mut head1))
        .await
        .expect("bind reply#1")
        .expect("read");
    assert_eq!(
        head1[1],
        consts::SOCKS5_REPLY_SUCCEEDED,
        "BIND through the door"
    );
    let mut bnd = [0u8; 6];
    timeout(BOUND, bind_stream.read_exact(&mut bnd))
        .await
        .expect("reply#1 BND.ADDR")
        .expect("read");
    let advertised = SocketAddr::from((
        [bnd[0], bnd[1], bnd[2], bnd[3]],
        u16::from_be_bytes([bnd[4], bnd[5]]),
    ));
    assert!(
        advertised.port() != 0,
        "reply#1 carries the real listener: {advertised}"
    );
    let mut app = timeout(BOUND, TcpStream::connect(advertised))
        .await
        .expect("dial the advertised address")
        .expect("app connected");
    app.write_all(b"vanilla-bind").await.expect("app writes");
    {
        let mut head2 = [0u8; 4];
        timeout(BOUND, bind_stream.read_exact(&mut head2))
            .await
            .expect("bind reply#2 head")
            .expect("read");
        assert_eq!(head2[1], consts::SOCKS5_REPLY_SUCCEEDED, "BIND reply#2");
        let mut peer_addr = [0u8; 6];
        timeout(BOUND, bind_stream.read_exact(&mut peer_addr))
            .await
            .expect("reply#2 DST.ADDR")
            .expect("read");
        let port = u16::from_be_bytes([peer_addr[4], peer_addr[5]]);
        assert_eq!(
            port,
            app.local_addr().expect("app local").port(),
            "reply#2 carries the real peer"
        );
    }

    // --- ASSOCIATE through the UDP door (live, mid-life of both) ----
    let client_sock = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    client_sock
        .send_to(&rfc_datagram(udp_target, b"udp-door"), udp_relay_addr)
        .await
        .expect("vanilla sendto the relay");
    let mut rx = vec![0u8; 65535];
    let (n, from) = timeout(BOUND, client_sock.recv_from(&mut rx))
        .await
        .expect("udp reply within bound")
        .expect("recv");
    assert_eq!(
        from, udp_relay_addr,
        "the reply comes from the relay socket"
    );
    let (src, payload) = parse_rfc_datagram(&rx[..n]);
    assert_eq!(src, udp_target, "the echo reports the target");
    assert_eq!(payload, b"udp-door");

    // --- The data planes, all three live: exercise each mid-life ----
    connect_stream
        .write_all(b"tcp-live")
        .await
        .expect("tcp write");
    let mut buf = [0u8; 8];
    timeout(BOUND, connect_stream.read_exact(&mut buf))
        .await
        .expect("tcp echo")
        .expect("read");
    assert_eq!(&buf, b"tcp-live");

    let mut app_buf = [0u8; 12];
    timeout(BOUND, bind_stream.read_exact(&mut app_buf))
        .await
        .expect("bind data plane")
        .expect("read");
    assert_eq!(&app_buf, b"vanilla-bind");

    client_sock
        .send_to(&rfc_datagram(udp_target, b"udp-live"), udp_relay_addr)
        .await
        .expect("send");
    let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx))
        .await
        .expect("udp echo")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"udp-live");

    // --- Teardown: the door-client disconnect ends only its own -----
    // The CONNECT + BIND connections drop (the vanilla client's
    // disconnect); the association on the UDP door is a separate
    // session.
    drop(connect_stream);
    drop(bind_stream);
    drop(app);

    // The UDP door's association survives the door-client disconnect
    // (mux/demux isolation): another datagram round-trips.
    client_sock
        .send_to(&rfc_datagram(udp_target, b"still"), udp_relay_addr)
        .await
        .expect("send after the tcp teardown");
    let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx))
        .await
        .expect("post-teardown echo")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"still");

    // Fresh sessions of every command succeed immediately after (no
    // connection poisoning).
    let (mut fresh, code, _) = timeout(BOUND, vanilla_cred_connect(door_addr, tcp_target))
        .await
        .expect("fresh CONNECT through the door");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
    fresh.write_all(b"again").await.expect("write");
    let mut buf = [0u8; 5];
    timeout(BOUND, fresh.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"again");

    client_sock
        .send_to(&rfc_datagram(udp_target, b"fresh"), udp_relay_addr)
        .await
        .expect("send fresh");
    let (n, _) = timeout(BOUND, client_sock.recv_from(&mut rx))
        .await
        .expect("fresh udp echo")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"fresh");
}

/// The teardown matrix, door edition (the no-leak witness): the door
/// client's disconnect tears the association's channel down (the
/// channel count witnesses); a fresh association succeeds immediately
/// after (no connection poisoning).
#[tokio::test]
async fn door_client_disconnect_tears_the_association_down_no_leaks() {
    init_trace();
    let udp_target = echo_udp_server().await;
    let topo = wire_consumer_side_with(
        alksocks::local::tcp_dial(),
        Some(alksocks::local::udp_egress()),
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move { door.serve_on(relay, client).await.expect("the door serves") }
    });

    // The association is live: a datagram round-trips and the channel
    // exists.
    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
    client
        .send_to(&rfc_datagram(udp_target, b"pre"), relay_addr)
        .await
        .expect("send");
    let mut rx = vec![0u8; 65535];
    let (n, _) = timeout(BOUND, client.recv_from(&mut rx))
        .await
        .expect("reply")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"pre");
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .any(|&id| id != 0),
        "the association rode a channel"
    );

    // The door client vanishes: the idle bound (the constructor knob)
    // is the door's teardown trigger for a silent relay: with the
    // socket gone, no datagrams arrive, the bound fires, the
    // association ends (association lifetime = channel lifetime,
    // ADR-003), and the door's task returns.
    drop(client);
    timeout(Duration::from_secs(35), serve)
        .await
        .expect("the door's task ended on the idle bound")
        .expect("clean exit");

    // No leak: the association's channel reaped.
    tokio::time::sleep(Duration::from_millis(200)).await;
    assert!(
        topo.consumer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0),
        "the teardown reaped the association's channel: {:?}",
        topo.consumer.manager().channel_ids()
    );

    // Fresh association succeeds immediately (no connection
    // poisoning).
    let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
    let relay2_addr = relay2.local_addr().expect("relay addr 2");
    let door2 = UdpFrontDoor::builder()
        .bind(relay2_addr)
        .build()
        .expect("the door builds");
    let _serve2 = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move {
            door2
                .serve_on(relay2, client)
                .await
                .expect("the fresh door serves")
        }
    });
    let client2 = UdpSocket::bind("127.0.0.1:0").await.expect("client 2");
    client2
        .send_to(&rfc_datagram(udp_target, b"fresh"), relay2_addr)
        .await
        .expect("send");
    let (n, _) = timeout(BOUND, client2.recv_from(&mut rx))
        .await
        .expect("fresh reply")
        .expect("recv");
    assert_eq!(parse_rfc_datagram(&rx[..n]).1, b"fresh");
}

/// The composed story end to end (the ADR-008 composed variant set,
/// all three commands): the SOCKS5 producer serves
/// `composed::direct_egress` (ASSOCIATE, the lazy destination→tunnel
/// table) + `composed::bindfirst_accept` (BIND, the bind-first
/// listener) + a real TCP dial (CONNECT). The channels-native
/// sessions drive all three against the one registration; the
/// direct-op dial count witnesses the table's lazy shape.
#[tokio::test]
async fn composed_story_all_three_commands_on_one_registration() {
    let suite = wire_composed_suite().await;

    // CONNECT through the dial seam (a real TCP echo target,
    // dialed by the producer's local dial).
    let tcp_target = echo_tcp_server().await;
    let mut session = timeout(
        BOUND,
        Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)),
    )
    .await
    .expect("connect within bound")
    .expect("the CONNECT session");
    let (mut read, mut write) = session.split().expect("pipe");
    write.write_all(b"composed-connect").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = [0u8; 16];
    timeout(BOUND, read.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"composed-connect");

    // ASSOCIATE through the composed egress (the direct op): one
    // destination → one tunnel (the dial count witnesses the lazy
    // open), the framed datagram round-trips through it.
    let mut assoc = timeout(
        BOUND,
        AssociateSession::associate(suite.topo.opener.as_ref()),
    )
    .await
    .expect("associate within bound")
    .expect("the association");
    let dest = TargetAddr::Ip(SocketAddr::from(([10, 9, 9, 9], 5353)));
    assoc
        .send_datagram(&dest, b"composed-udp")
        .await
        .expect("send");
    let (src, payload) = timeout(BOUND, assoc.recv_datagram())
        .await
        .expect("recv within bound")
        .expect("a datagram")
        .expect("a datagram (not eof)");
    assert_eq!(src, dest, "the echo reports the destination");
    assert_eq!(&payload[..], b"composed-udp");
    assert_eq!(
        suite.dial_count.load(std::sync::atomic::Ordering::SeqCst),
        1,
        "one direct-op dial for the one destination"
    );

    // BIND through the composed accept (the bind-first listener):
    // reply#1 = the real bound; the queue accept; reply#2 = the real
    // peer; the pass-through data plane.
    let mut bind = timeout(
        BOUND,
        BindSession::bind(
            suite.topo.opener.as_ref(),
            ("bind.test", 2121).to_target_addr().unwrap(),
        ),
    )
    .await
    .expect("bind within bound")
    .expect("the BIND session");
    let bind_addr = bind.reply1().expect("reply#1").bind_addr;
    assert_eq!(
        bind_addr,
        TargetAddr::Ip(SocketAddr::from(([127, 0, 0, 1], COMPOSED_BOUND_PORT))),
        "reply#1 advertises the real OS-chosen bound"
    );
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    let mut plane = timeout(BOUND, bind.await_inbound())
        .await
        .expect("reply#2")
        .expect("await_inbound ok")
        .expect("the plane");
    assert_eq!(
        plane.peer,
        TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], COMPOSED_PEER_PORT))),
        "reply#2 carries the forwarded op's real peer"
    );
    plane.write_all(b"composed-bind").await.expect("write");
    plane.flush().await.expect("flush");
    let mut bbuf = [0u8; 13];
    timeout(BOUND, plane.read_exact(&mut bbuf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&bbuf, b"composed-bind");

    // All three sessions were live simultaneously; the teardowns end
    // only their own sessions (mux/demux isolation), exercised by
    // the interleaving above. Drop + close.
    drop(read);
    drop(write);
    drop(assoc);
    drop(plane);
    assert!(session.close().await, "the CONNECT channel reaped");
    assert!(bind.close().await, "the BIND channel reaped");
}

/// Mixed assembly (the producer/consumer-roles-not-sides posture): one
/// producer registration serving channels-native sessions AND door
/// clients AND composed seams concurrently. The composed suite's
/// producer gains a door-shaped second consumer topology over the same
/// registration. The channels-native sessions and the door clients
/// interleave on the one registration (both shapes, one connection).
#[tokio::test]
async fn mixed_assembly_channels_native_and_door_clients_concurrent() {
    init_trace();
    let tcp_target = echo_tcp_server().await;
    let udp_target = echo_udp_server().await;

    let topo = wire_consumer_side_with_provider(
        alksocks::local::tcp_dial(),
        Some(alksocks::local::udp_egress()),
        Some(alksocks::local::bind_accept([127, 0, 0, 1])),
        Some(harness::consumer_identity()),
        Arc::new(DoorTokenProvider {
            bind_scoped: vec!["door-mixed".to_string()],
        }),
    )
    .await;
    let door_topo = wire_consumer_side_with_provider(
        alksocks::local::tcp_dial(),
        Some(alksocks::local::udp_egress()),
        Some(alksocks::local::bind_accept([127, 0, 0, 1])),
        Some(harness::consumer_identity()),
        Arc::new(DoorTokenProvider {
            bind_scoped: vec!["door-mixed".to_string()],
        }),
    )
    .await;
    let udp_door_topo = wire_consumer_side_with_provider(
        alksocks::local::tcp_dial(),
        Some(alksocks::local::udp_egress()),
        Some(alksocks::local::bind_accept([127, 0, 0, 1])),
        Some(harness::consumer_identity()),
        Arc::new(DoorTokenProvider {
            bind_scoped: vec!["door-mixed".to_string()],
        }),
    )
    .await;

    // The door fronts (the vanilla shape) on their own connections to
    // the same registration (the cred-bearing door's TokenClient takes
    // its client's call connection: the door-only contract; the UDP
    // door's association rides a second connection).
    let tcp_listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let door_addr = tcp_listener.local_addr().expect("door addr");
    let door = FrontDoor::builder()
        .credentials(cred_table())
        .bind(door_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&door_topo.consumer);
        async move {
            door.serve_on(tcp_listener, client)
                .await
                .expect("the door serves")
        }
    });
    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
    let relay_addr = relay.local_addr().expect("relay addr");
    let udp_door = UdpFrontDoor::builder()
        .bind(relay_addr)
        .build()
        .expect("the door builds");
    let _udp_serve = tokio::spawn({
        let client = Arc::clone(&udp_door_topo.consumer);
        async move {
            udp_door
                .serve_on(relay, client)
                .await
                .expect("the udp door serves")
        }
    });

    // The three shapes, concurrent on the one registration:
    //
    // 1. A channels-native CONNECT + ASSOCIATE + BIND session trio
    //    (the wrapper-aware shape).
    // 2. A door-client CONNECT round-trip (the vanilla RFC 1929
    //    shape through the cred-bearing door).
    // 3. A door-client ASSOCIATE datagram (the vanilla RFC-header'd
    //    shape through the UDP door).
    //
    // All six run interleaved (join_all), then each is verified
    // mid-life of the others.

    let native = async {
        // CONNECT native.
        let mut s =
            Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(tcp_target))
                .await
                .map_err(|e| format!("native connect: {e}"))?;
        let (mut read, mut write) = s.split().ok_or_else(|| "no pipe".to_string())?;
        write
            .write_all(b"native-tcp")
            .await
            .map_err(|e| e.to_string())?;
        write.flush().await.map_err(|e| e.to_string())?;
        let mut buf = [0u8; 10];
        read.read_exact(&mut buf).await.map_err(|e| e.to_string())?;
        if &buf != b"native-tcp" {
            return Err("native connect echo mismatch".to_string());
        }
        drop(read);
        drop(write);
        s.close().await;
        Ok(())
    };

    let native_udp = async {
        let mut a = AssociateSession::associate(topo.consumer.as_ref())
            .await
            .map_err(|e| format!("native associate: {e}"))?;
        let target = TargetAddr::Ip(udp_target);
        a.send_datagram(&target, b"native-udp")
            .await
            .map_err(|e| e.to_string())?;
        let (src, payload) = a
            .recv_datagram()
            .await
            .map_err(|e| e.to_string())?
            .ok_or_else(|| "eof before a datagram".to_string())?;
        if src != target || &payload[..] != b"native-udp" {
            return Err("native associate echo mismatch".to_string());
        }
        a.close().await;
        Ok(())
    };

    let door_tcp = async {
        let mut stream = timeout(BOUND, vanilla_with_creds(door_addr))
            .await
            .map_err(|e| format!("door connect: {e}"))?;
        let mut req = vec![0x05, 0x01, 0x00, 0x01];
        let SocketAddr::V4(v4) = tcp_target else {
            panic!("v4")
        };
        req.extend_from_slice(&v4.ip().octets());
        req.extend_from_slice(&v4.port().to_be_bytes());
        stream.write_all(&req).await.map_err(|e| e.to_string())?;
        let mut head = [0u8; 4];
        timeout(BOUND, stream.read_exact(&mut head))
            .await
            .map_err(|e| e.to_string())?
            .map_err(|e| e.to_string())?;
        if head[1] != consts::SOCKS5_REPLY_SUCCEEDED {
            return Err(format!("door connect reply {}", head[1]));
        }
        let mut bnd = [0u8; 6];
        timeout(BOUND, stream.read_exact(&mut bnd))
            .await
            .map_err(|e| e.to_string())?
            .map_err(|e| e.to_string())?;
        stream
            .write_all(b"door-tcp")
            .await
            .map_err(|e| e.to_string())?;
        let mut buf = [0u8; 8];
        timeout(BOUND, stream.read_exact(&mut buf))
            .await
            .map_err(|e| e.to_string())?
            .map_err(|e| e.to_string())?;
        if &buf != b"door-tcp" {
            return Err("door connect echo mismatch".to_string());
        }
        Ok(())
    };

    let door_udp = async {
        let sock = UdpSocket::bind("127.0.0.1:0")
            .await
            .map_err(|e| e.to_string())?;
        sock.send_to(&rfc_datagram(udp_target, b"door-udp"), relay_addr)
            .await
            .map_err(|e| e.to_string())?;
        let mut rx = vec![0u8; 65535];
        let (n, _) = timeout(BOUND, sock.recv_from(&mut rx))
            .await
            .map_err(|e| format!("door udp recv: {e}"))?
            .map_err(|e| e.to_string())?;
        if parse_rfc_datagram(&rx[..n]).1 != b"door-udp" {
            return Err("door udp echo mismatch".to_string());
        }
        Ok(())
    };

    let (r1, r2, r3, r4) = tokio::join!(native, native_udp, door_tcp, door_udp);
    r1.expect("native connect");
    r2.expect("native associate");
    r3.expect("door connect");
    r4.expect("door associate");

    // The registration is healthy after the mixed burst: one more of
    // each shape succeeds.
    let (mut fresh, code, _) = timeout(BOUND, vanilla_cred_connect(door_addr, tcp_target))
        .await
        .expect("fresh door connect");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
    fresh.write_all(b"again").await.expect("write");
    let mut buf = [0u8; 5];
    timeout(BOUND, fresh.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"again");

    let mut fresh_native =
        Socks5ChannelSession::connect(topo.consumer.as_ref(), TargetAddr::Ip(tcp_target))
            .await
            .expect("fresh native connect");
    let (mut read, mut write) = fresh_native.split().expect("pipe");
    write.write_all(b"again2").await.expect("write");
    let mut buf2 = [0u8; 6];
    timeout(BOUND, read.read_exact(&mut buf2))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf2, b"again2");
    drop(read);
    drop(write);
    fresh_native.close().await;
}

/// The mixed-composed variant of the mixed-assembly pin: channels-
/// native sessions AND composed seams AND door clients on one
/// registration: the composed suite's producer + a door consumer
/// topology, both shapes live on one connection (roles-not-sides: the
/// producer serves every shape through its seams; the connection
/// direction is independent of the service direction).
#[tokio::test]
async fn mixed_assembly_composed_seams_beside_channels_native_sessions() {
    let suite = wire_composed_suite().await;
    let tcp_target = echo_tcp_server().await;

    // Channels-native CONNECT + composed BIND + composed ASSOCIATE,
    // all opened before any is exercised (live simultaneously), then
    // each exercised mid-life of the others; teardowns end only their
    // own sessions.
    let mut connect = timeout(
        BOUND,
        Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)),
    )
    .await
    .expect("connect")
    .expect("the CONNECT session");
    let mut assoc = timeout(
        BOUND,
        AssociateSession::associate(suite.topo.opener.as_ref()),
    )
    .await
    .expect("associate")
    .expect("the association");
    let mut bind = timeout(
        BOUND,
        BindSession::bind(
            suite.topo.opener.as_ref(),
            ("bind.test", 2121).to_target_addr().unwrap(),
        ),
    )
    .await
    .expect("bind")
    .expect("the BIND session");

    // The BIND's quiet wait is live (the listener exists); complete it
    // mid-life.
    composed_push_accept(&suite.queue, ("10.0.0.1", COMPOSED_PEER_PORT)).await;
    let mut plane = timeout(BOUND, bind.await_inbound())
        .await
        .expect("reply#2")
        .expect("await_inbound ok")
        .expect("the plane");
    plane.write_all(b"mid").await.expect("write");
    let mut bbuf = [0u8; 3];
    timeout(BOUND, plane.read_exact(&mut bbuf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&bbuf, b"mid");

    // The CONNECT + ASSOCIATE sessions still serve (no connection
    // poisoning).
    let (mut read, mut write) = connect.split().expect("pipe");
    write.write_all(b"alive").await.expect("write");
    let mut buf = [0u8; 5];
    timeout(BOUND, read.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"alive");

    let dest = TargetAddr::Ip(SocketAddr::from(([10, 9, 9, 9], 5353)));
    assoc.send_datagram(&dest, b"alive").await.expect("send");
    let (src, payload) = timeout(BOUND, assoc.recv_datagram())
        .await
        .expect("recv")
        .expect("a datagram")
        .expect("a datagram (not eof)");
    assert_eq!(src, dest);
    assert_eq!(&payload[..], b"alive");

    // Teardowns end only their own sessions; the registration serves
    // a fresh CONNECT after.
    drop(read);
    drop(write);
    drop(assoc);
    drop(plane);
    assert!(connect.close().await);
    assert!(bind.close().await);
    let mut fresh = timeout(
        BOUND,
        Socks5ChannelSession::connect(suite.topo.opener.as_ref(), TargetAddr::Ip(tcp_target)),
    )
    .await
    .expect("fresh connect")
    .expect("the fresh session");
    let (mut read, mut write) = fresh.split().expect("pipe");
    write.write_all(b"again").await.expect("write");
    let mut buf = [0u8; 5];
    timeout(BOUND, read.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"again");
    drop(read);
    drop(write);
    fresh.close().await;

    // The handler-exit teardowns reap asynchronously (the open
    // wrapper's post-handler teardown); settle past the race, then
    // probe (the composed suites' shape).
    for _ in 0..50 {
        tokio::time::sleep(Duration::from_millis(100)).await;
        let ids = suite.topo.producer_manager.channel_ids();
        if ids.iter().all(|&id| id == 0 || id % 2 == 1) {
            break;
        }
    }
    // No channel leaks (the composed suites' probe; the producer side
    // may retain the adopted forwarded entries: the documented
    // concession).
    harness::assert_composed_channels(&suite.topo, &[0], &[]).await;
}

/// The producer-side local backend as the `-D` story's other half (the
/// one-registration-three-commands regression through the producer's
/// own front door): a vanilla NoAuth client drives CONNECT + BIND +
/// ASSOCIATE through one `ProducerDoor` against one producer
/// registration (the `local` provider set), all three live
/// simultaneously, one accepted connection per command (the RFC
/// conversation is one command per accepted socket, the vanilla
/// server's shape).
#[tokio::test]
async fn producer_door_one_registration_three_commands() {
    init_trace();
    let tcp_target = echo_tcp_server().await;
    let udp_target = echo_udp_server().await;

    let producer = alksocks::producer::Socks5Producer::builder()
        .dial(alksocks::local::tcp_dial())
        .egress(alksocks::local::udp_egress())
        .accept(alksocks::local::bind_accept([127, 0, 0, 1]))
        .build();
    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let door_addr = listener.local_addr().expect("door addr");
    let door = alksocks::local::ProducerDoor::builder()
        .producer(producer)
        .build()
        .expect("the door builds");
    let _guard = door.serve_on(listener).await;

    // CONNECT (live).
    let (mut connect_stream, code, _) =
        timeout(BOUND, vanilla_no_auth_connect(door_addr, tcp_target))
            .await
            .expect("connect round trip");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);

    // BIND (live, mid-life of the CONNECT): its own door connection.
    let mut bind_stream = TcpStream::connect(door_addr).await.expect("door connect");
    bind_stream
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    timeout(BOUND, bind_stream.read_exact(&mut chosen))
        .await
        .expect("method selection")
        .expect("read");
    assert_eq!(chosen, [0x05, 0x00]);
    bind_stream
        .write_all(&[0x05, 0x02, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("bind request");
    let mut head1 = [0u8; 4];
    timeout(BOUND, bind_stream.read_exact(&mut head1))
        .await
        .expect("bind reply#1")
        .expect("read");
    assert_eq!(head1[1], consts::SOCKS5_REPLY_SUCCEEDED);
    let mut bnd = [0u8; 6];
    timeout(BOUND, bind_stream.read_exact(&mut bnd))
        .await
        .expect("reply#1 BND.ADDR")
        .expect("read");
    let advertised = SocketAddr::from((
        [bnd[0], bnd[1], bnd[2], bnd[3]],
        u16::from_be_bytes([bnd[4], bnd[5]]),
    ));
    let mut app = timeout(BOUND, TcpStream::connect(advertised))
        .await
        .expect("dial the advertised address")
        .expect("app connected");
    app.write_all(b"prod-bind").await.expect("app writes");
    let mut head2 = [0u8; 4];
    timeout(BOUND, bind_stream.read_exact(&mut head2))
        .await
        .expect("bind reply#2")
        .expect("read");
    assert_eq!(head2[1], consts::SOCKS5_REPLY_SUCCEEDED, "BIND reply#2");
    let mut peer_addr = [0u8; 6];
    timeout(BOUND, bind_stream.read_exact(&mut peer_addr))
        .await
        .expect("reply#2 DST.ADDR")
        .expect("read");
    assert_eq!(
        u16::from_be_bytes([peer_addr[4], peer_addr[5]]),
        app.local_addr().expect("app local").port(),
        "reply#2 carries the real peer"
    );

    // ASSOCIATE (live, mid-life of both): its own door connection.
    // It gets the sentinel reply + the in-band framed datagram through
    // the real egress (the producer door path speaks the in-band wire;
    // the ASSOCIATE datagram plane has no door-side RFC codec here;
    // the vanilla client is wrapper-aware on this path).
    let mut assoc_stream = TcpStream::connect(door_addr).await.expect("door connect");
    assoc_stream
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen3 = [0u8; 2];
    timeout(BOUND, assoc_stream.read_exact(&mut chosen3))
        .await
        .expect("method selection")
        .expect("read");
    assert_eq!(chosen3, [0x05, 0x00]);
    assoc_stream
        .write_all(&[0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0])
        .await
        .expect("associate request");
    let mut head3 = [0u8; 4];
    timeout(BOUND, assoc_stream.read_exact(&mut head3))
        .await
        .expect("associate reply")
        .expect("read");
    assert_eq!(
        head3[1],
        consts::SOCKS5_REPLY_SUCCEEDED,
        "ASSOCIATE through the door"
    );
    let mut bnd3 = [0u8; 6];
    timeout(BOUND, assoc_stream.read_exact(&mut bnd3))
        .await
        .expect("the sentinel BND.ADDR")
        .expect("read");
    assert_eq!(&bnd3, &[127, 0, 0, 1, 0, 0], "the sentinel");

    // All three data planes, live simultaneously.
    connect_stream.write_all(b"prod-tcp").await.expect("write");
    let mut buf = [0u8; 8];
    timeout(BOUND, connect_stream.read_exact(&mut buf))
        .await
        .expect("tcp echo")
        .expect("read");
    assert_eq!(&buf, b"prod-tcp");

    let mut app_buf = [0u8; 9];
    timeout(BOUND, bind_stream.read_exact(&mut app_buf))
        .await
        .expect("bind data plane")
        .expect("read");
    assert_eq!(&app_buf, b"prod-bind");

    let dst = TargetAddr::Ip(udp_target);
    let frame = alksocks::wire::frame(&dst, b"prod-udp").expect("frame");
    assoc_stream.write_all(&frame).await.expect("datagram");
    let mut reply = vec![0u8; frame.len()];
    timeout(BOUND, assoc_stream.read_exact(&mut reply))
        .await
        .expect("udp echo")
        .expect("read");
    let n = u16::from_be_bytes([reply[0], reply[1]]) as usize;
    let inner = &reply[2..2 + n];
    let atyp = inner[0];
    let addr_len = match atyp {
        0x01 => 1 + 4 + 2,
        0x03 => 1 + 1 + inner[1] as usize + 2,
        0x04 => 1 + 16 + 2,
        other => panic!("unexpected ATYP {other}"),
    };
    assert_eq!(&inner[addr_len..], b"prod-udp", "the datagram echoed");

    // The teardown: the client drops; the door stays serving (the
    // loop continues, the per-accept isolation); a fresh session of
    // every command succeeds (fresh CONNECT here; the door is
    // healthy).
    drop(connect_stream);
    drop(app);
    let (_, code, _) = timeout(BOUND, vanilla_no_auth_connect(door_addr, tcp_target))
        .await
        .expect("fresh connect after the teardown");
    assert_eq!(
        code,
        consts::SOCKS5_REPLY_SUCCEEDED,
        "no connection poisoning"
    );
}

/// The `-D` story's vanilla-client-fidelity pin (tun2proxy-shaped): a
/// domain-form CONNECT through the cred-bearing door resolves at the
/// producer's seam (the ssh `-D` semantic, ADR-012); `localhost` is
/// reachable as the echo target.
#[tokio::test]
async fn dash_d_domain_targets_resolve_at_the_producer_seam() {
    init_trace();
    let tcp_target = echo_tcp_server().await;
    let topo = wire_consumer_side_with(
        alksocks::local::tcp_dial(),
        None,
        None,
        Some(harness::consumer_identity()),
    )
    .await;

    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
    let door_addr = listener.local_addr().expect("door addr");
    let door = FrontDoor::builder()
        .bind(door_addr)
        .build()
        .expect("the door builds");
    let _serve = tokio::spawn({
        let client = Arc::clone(&topo.consumer);
        async move {
            door.serve_on(listener, client)
                .await
                .expect("the door serves")
        }
    });

    // The NoAuth door is the raw pipe: the vanilla client's whole RFC
    // conversation (domain-form CONNECT included) crosses it verbatim
    // and resolves at the producer's dial seam.
    let mut stream = TcpStream::connect(door_addr).await.expect("door connect");
    stream
        .write_all(&[0x05, 0x01, 0x00])
        .await
        .expect("greeting");
    let mut chosen = [0u8; 2];
    stream
        .read_exact(&mut chosen)
        .await
        .expect("method selection");
    assert_eq!(chosen, [0x05, 0x00]);

    let domain = "localhost".to_owned();
    let mut req = vec![0x05, 0x01, 0x00, 0x03, domain.len() as u8];
    req.extend_from_slice(domain.as_bytes());
    req.extend_from_slice(&tcp_target.port().to_be_bytes());
    stream.write_all(&req).await.expect("request");
    let mut head = [0u8; 4];
    timeout(BOUND, stream.read_exact(&mut head))
        .await
        .expect("reply")
        .expect("read");
    assert_eq!(
        head[1],
        consts::SOCKS5_REPLY_SUCCEEDED,
        "the domain resolved"
    );
    let mut bnd = [0u8; 6];
    timeout(BOUND, stream.read_exact(&mut bnd))
        .await
        .expect("BND.ADDR")
        .expect("read");
    stream.write_all(b"dom").await.expect("write");
    let mut buf = [0u8; 3];
    timeout(BOUND, stream.read_exact(&mut buf))
        .await
        .expect("echo")
        .expect("read");
    assert_eq!(&buf, b"dom");
}