local: the socket-backed seam providers — tcp_dial / udp_egress / bind_accept

The `local` feature's first-party providers (ADR-008's local variants,
ADR-009's gate): the three seams' faithful-RFC assembly-layer
constructors over the landed seam types; the protocol layer never
sees them.

- Cargo.toml: `local = ["tokio/net"]` (no socket2 — the POCs ran
  tokio-only); the default graph stays net/mio/socket2-free
  (cargo tree spot-checked both ways)
- src/local/: `tcp_dial()` (real TcpStream per CONNECT, domain
  resolution at the provider — the ssh -D semantic; the resolved
  address set iterated, faithful SocksDialError mapping,
  NotAllowed never synthesized), `udp_egress()` (one real UdpSocket
  per association bound 127.0.0.1:0 — the explicit bind decision;
  the POC #2 LocalEgress readiness shape: park on readable(),
  try_recv_from, WouldBlock -> spurious Ok(None); inner bytes
  to_be_bytes() + data, framing-once; resolve at the egress),
  `bind_accept(bind_host)` (real TcpListener per BIND session,
  ADR-007 variant 1: requested port honored when non-zero, local_addr
  as reply#1's BND.ADDR, the accept future owning the listener
  end-to-end, peer byte-exact; two-stage failure mapping; the bind
  host is the caller's explicit choice)
- lib.rs: cfg-gated module + re-exports; crate doc updated
- tests: 12 unit (real-socket round-trips, the readiness contract's
  park-then-datagram, the drop-closes-the-listener witness, the
  refusal mappings) + 6 integration (tests/local-providers.rs,
  feature-gated: the three commands end to end through the producer's
  real dispatch path with real targets, connect-refused -> 0x05
  through the producer's reply, the readiness contract re-pinned
  through the datagram stage, client-gone -> the advertised port
  frees)
- task status -> completed (acceptance criteria checked; Notes +
  Summary filled)

Verification: cargo test (157 green) + --all-features green,
clippy -D warnings clean (default + all-features), fmt --check clean,
both wasm checks clean, the dependency-graph spot-checks green.

src/lib.rs

@@ -42,10 +42,11 @@
 //!   [`BindSession`] (the reply#1/reply#2 reader) — each owning its
 //!   channel's teardown; no `Clone`.
 //!
-//! `local` and `composed` (the feature-gated assembly-layer
-//! providers: the front doors, the socket-backed/composed seams) are
-//! not yet present — the default crate is protocol-only and
-//! wasm-clean by construction (ADR-009).
+//! `local` (the socket-backed assembly-layer providers —
+//! [`local::tcp_dial`] / [`local::udp_egress`] / [`local::bind_accept`],
+//! ADR-008/009) is feature-gated: the default crate is protocol-only and
+//! wasm-clean by construction, and the composed providers (`composed`)
+//! are still pending.
 //!
 //! Architecture: `docs/architecture/overview.md` (spec set:
 //! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting
@@ -64,6 +65,9 @@ pub mod params;
 pub mod producer;
 pub mod wire;
 
+#[cfg(feature = "local")]
+pub mod local;
+
 /// The shared test helpers (review 002 N-5): the single-sourced
 /// harness shapes, compiled into the lib test crate. The file lives
 /// under `tests/common/` so the integration suite reaches the same
@@ -102,3 +106,6 @@ pub use producer::{
     socks5_session_establisher, IdentityWitness, Socks5Producer, Socks5ProducerBuilder,
 };
 pub use wire::{addr_encoding_len, frame, frame_inner, DatagramReader, WireError, MAX_INNER_LEN};
+
+#[cfg(feature = "local")]
+pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress};

src/local/mod.rs

@@ -0,0 +1,282 @@
+//! The `local` feature's socket-backed seam providers (ADR-008's local
+//! variants, ADR-009's feature gate): the three seams' faithful-RFC
+//! assembly-layer constructors — [`tcp_dial`] (CONNECT), [`udp_egress`]
+//! (UDP ASSOCIATE), [`bind_accept`] (TCP BIND). These are convenience
+//! constructors over the landed seam types ([`crate::egress`]); the
+//! protocol layer never sees them, and **the providers ARE the explicit
+//! bind decisions** — the never-binds posture (convention 9) holds
+//! because the protocol layer never references this module: every bind
+//! below (the dial's connect, the egress socket's `127.0.0.1:0`, the
+//! BIND listener) is the caller's explicit choice to enable this
+//! feature and wire the constructor.
+//!
+//! The seam contracts each constructor honors (the landed types'
+//! normative doc comments, `crate::egress`):
+//!
+//! - **Ownership rides the handle structurally** — the payloads own
+//!   their socket halves; dropping the handle drops the connection.
+//! - **The egress speaks inner bytes** — [`Egress::send`] takes the
+//!   payload only; [`Egress::next_reply`] returns
+//!   `TargetAddr::to_be_bytes() + data` (the *sender's* address, the
+//!   RFC's per-datagram source semantics); framing-once — the length
+//!   prefix is the wire layer's ([`crate::wire`]), added exactly once
+//!   downstream.
+//! - **The readiness contract** ([`Egress::next_reply`]) — parks on
+//!   `readable()` (a real async wait on the socket's receive
+//!   readiness), then `try_recv_from`; `WouldBlock` → the
+//!   spurious-wakeup `Ok(None)` shape; never a hot `Ok(None)`.
+//! - **The accept future owns its resources end-to-end** — the
+//!   listener lives inside the accept future; the quiet wait's drop IS
+//!   the client-gone teardown.
+//!
+//! Resolution is the provider's job (ADR-012): domain targets/destinations
+//! resolve here, at the seam — the ssh `-D` semantic.
+
+use std::io::ErrorKind;
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::{TcpListener, TcpStream, UdpSocket};
+
+pub use crate::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
+use crate::{
+    BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle,
+    DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle,
+};
+
+/// Resolve a [`TargetAddr`] to a `SocketAddr` (egress-side DNS; the
+/// ssh `-D` semantic — ADR-012: resolution is the provider's job). IP
+/// targets resolve directly; domain targets resolve via the blocking
+/// resolver (`std::net::ToSocketAddrs` — the POC #2 `LocalEgress`
+/// shape; the first address wins).
+pub(crate) fn resolve_target(addr: &TargetAddr) -> Result<SocketAddr, EgressError> {
+    match addr {
+        TargetAddr::Ip(sa) => Ok(*sa),
+        TargetAddr::Domain(domain, port) => {
+            use std::net::ToSocketAddrs;
+            (domain.as_str(), *port)
+                .to_socket_addrs()
+                .map_err(|e| EgressError::Unreachable(format!("resolve {domain}: {e}")))?
+                .next()
+                .ok_or_else(|| EgressError::Unreachable(format!("resolve {domain}: no addresses")))
+        }
+    }
+}
+
+/// Map an io error kind to the dial vocabulary (faithful-RFC mapping;
+/// `NotAllowed` is never synthesized by this provider — it is a
+/// ruleset concept, not an io one).
+pub(crate) fn dial_error_of(what: &str, err: &std::io::Error) -> SocksDialError {
+    match err.kind() {
+        ErrorKind::ConnectionRefused => SocksDialError::Refused(format!("{what}: {err}")),
+        ErrorKind::NetworkUnreachable => {
+            SocksDialError::NetworkUnreachable(format!("{what}: {err}"))
+        }
+        ErrorKind::HostUnreachable => SocksDialError::HostUnreachable(format!("{what}: {err}")),
+        ErrorKind::TimedOut => SocksDialError::NetworkUnreachable(format!("{what}: {err}")),
+        ErrorKind::PermissionDenied => SocksDialError::General(format!("{what}: {err}")),
+        _ => SocksDialError::General(format!("{what}: {err}")),
+    }
+}
+
+/// The CONNECT dial seam's local-socket variant (ADR-008): a real
+/// `TcpStream` per CONNECT target; domain targets resolve here (the
+/// ssh `-D` semantic). The connected stream splits into the
+/// [`TargetHandle`] halves (`TcpStream::into_split` — the halves are
+/// `Send + Sync + Unpin`, boxed per the seam's shape). Dial failures
+/// map to [`SocksDialError`] faithfully ([`dial_error_of`]);
+/// `NotAllowed` is never synthesized by this provider.
+pub fn tcp_dial() -> DialFn {
+    Arc::new(|target: TargetAddr| Box::pin(async move { local_tcp_dial(target).await }))
+}
+
+async fn local_tcp_dial(target: TargetAddr) -> Result<TargetHandle, SocksDialError> {
+    let addrs: Vec<SocketAddr> = match &target {
+        TargetAddr::Ip(sa) => vec![*sa],
+        TargetAddr::Domain(domain, port) => {
+            use std::net::ToSocketAddrs;
+            (domain.as_str(), *port)
+                .to_socket_addrs()
+                .map_err(|e| SocksDialError::HostUnreachable(format!("resolve {domain}: {e}")))?
+                .collect()
+        }
+    };
+    if addrs.is_empty() {
+        return Err(SocksDialError::HostUnreachable(format!(
+            "resolve {target}: no addresses"
+        )));
+    }
+    let mut last: Option<SocksDialError> = None;
+    for addr in addrs {
+        match TcpStream::connect(addr).await {
+            Ok(stream) => {
+                let (read, write) = stream.into_split();
+                return Ok(TargetHandle {
+                    read: Box::new(read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
+                    write: Box::new(write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
+                });
+            }
+            Err(err) => last = Some(dial_error_of(&format!("connect {target}"), &err)),
+        }
+    }
+    Err(last.unwrap_or_else(|| {
+        SocksDialError::General(format!("connect {target}: no addresses tried"))
+    }))
+}
+
+/// The ASSOCIATE egress seam's local-socket variant (ADR-008): one
+/// real `UdpSocket` per association, bound `127.0.0.1:0` (the
+/// assembly-layer bind decision, explicit here); per-datagram
+/// `send_to`/`recv_from`, no flow table; domain destinations resolve
+/// at the egress ([`resolve_target`]).
+///
+/// **The readiness contract is load-bearing**: [`LocalEgress`]'s
+/// `next_reply` parks on `sock.readable().await` (a real async park on
+/// receive readiness — `writable()` is always ready for UDP and would
+/// spin), then `try_recv_from` (`WouldBlock` → the spurious-wakeup
+/// `Ok(None)` shape; a datagram arriving makes the socket readable).
+/// Replies carry *inner* bytes (`TargetAddr::to_be_bytes() + data` —
+/// the sender's address per the RFC's per-datagram source semantics;
+/// framing-once — the length prefix is the wire layer's). `send` takes
+/// the payload only.
+pub fn udp_egress() -> EgressFactory {
+    Arc::new(|| {
+        Box::pin(async {
+            LocalEgress::bind()
+                .await
+                .map(|e| Box::new(e) as Box<dyn Egress + Send>)
+        })
+    })
+}
+
+/// One real UDP socket per association (the POC #2 `LocalEgress`
+/// shape, ported). See [`udp_egress`] for the contracts.
+pub struct LocalEgress {
+    sock: Arc<UdpSocket>,
+}
+
+impl LocalEgress {
+    /// Bind the association's socket (the explicit bind decision:
+    /// loopback ephemeral — a wildcard bind is the caller's variant
+    /// to add, never this constructor's default).
+    pub async fn bind() -> Result<Self, EgressError> {
+        let sock = UdpSocket::bind("127.0.0.1:0")
+            .await
+            .map_err(|e| EgressError::General(format!("udp bind: {e}")))?;
+        Ok(Self {
+            sock: Arc::new(sock),
+        })
+    }
+}
+
+impl Egress for LocalEgress {
+    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> {
+        let sock = Arc::clone(&self.sock);
+        Box::pin(async move {
+            let dest = resolve_target(&addr)?;
+            sock.send_to(&payload, dest)
+                .await
+                .map(|_| ())
+                .map_err(|e| EgressError::General(format!("udp send_to {dest}: {e}")))
+        })
+    }
+
+    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+        let sock = Arc::clone(&self.sock);
+        Box::pin(async move {
+            // A real async wait on the socket's receive readiness (a
+            // datagram arriving makes it readable): the stage's
+            // select! parks here with the channel read — no busy-loop,
+            // no lost replies.
+            sock.readable()
+                .await
+                .map_err(|e| EgressError::General(format!("udp readable: {e}")))?;
+            let mut scratch = vec![0u8; 65535];
+            match sock.try_recv_from(&mut scratch) {
+                Ok((n, from)) => {
+                    // The INNER bytes (`addr + data` — the RFC's
+                    // per-datagram source semantics): the length
+                    // prefix is the wire layer's, applied exactly
+                    // once downstream.
+                    let mut inner = TargetAddr::Ip(from)
+                        .to_be_bytes()
+                        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
+                    inner.extend_from_slice(&scratch[..n]);
+                    Ok(Some(Bytes::from(inner)))
+                }
+                // A spurious wakeup (the readiness contract's
+                // tolerated shape — the stage re-parks; this happens
+                // async, never a hot loop).
+                Err(e) if e.kind() == ErrorKind::WouldBlock => Ok(None),
+                Err(e) => Err(EgressError::General(format!("udp try_recv: {e}"))),
+            }
+        })
+    }
+}
+
+/// The BIND accept seam's local-socket variant (ADR-008, the faithful
+/// RFC shape, ADR-007 variant 1): a real `TcpListener` per BIND
+/// session. The requested target's port is honored when non-zero (FTP
+/// active mode); port 0 binds an ephemeral port; `local_addr()` is
+/// reply#1's `bind_addr`. The accept future owns the listener
+/// end-to-end (the quiet wait's drop IS the client-gone teardown — the
+/// listener lives inside the future, never in a separate binding);
+/// `accept()`'s peer is reply#2's [`BindTargetHandle::peer`]
+/// (byte-exact). Factory failures map to [`BindAcceptFactoryError`]
+/// (bind refused/held port → `Refused`/`Shortage` per the io error);
+/// accept-stage failures → [`BindAcceptError`].
+///
+/// **Address policy is the caller's**: `bind_host` is taken
+/// explicitly — loopback (`127.0.0.1`) is the POC's posture; a
+/// wildcard bind is the caller's explicit choice, spelled here.
+pub fn bind_accept(bind_host: impl Into<std::net::IpAddr>) -> BindAcceptFactory {
+    let bind_host = bind_host.into();
+    Arc::new(move |target: TargetAddr| {
+        Box::pin(async move { local_bind_accept(bind_host, &target).await })
+    })
+}
+
+async fn local_bind_accept(
+    bind_host: std::net::IpAddr,
+    target: &TargetAddr,
+) -> Result<BindAccept, BindAcceptFactoryError> {
+    let requested_port = match target {
+        TargetAddr::Ip(addr) => addr.port(),
+        TargetAddr::Domain(_, port) => *port,
+    };
+    let listener = TcpListener::bind(SocketAddr::new(bind_host, requested_port))
+        .await
+        .map_err(|e| match e.kind() {
+            ErrorKind::PermissionDenied | ErrorKind::AddrInUse => {
+                BindAcceptFactoryError::Shortage(format!("tcp bind: {e}"))
+            }
+            ErrorKind::AddrNotAvailable => {
+                BindAcceptFactoryError::Refused(format!("tcp bind: {e}"))
+            }
+            _ => BindAcceptFactoryError::General(format!("tcp bind: {e}")),
+        })?;
+    let bind_addr = listener
+        .local_addr()
+        .map_err(|e| BindAcceptFactoryError::General(format!("local_addr: {e}")))?;
+    Ok(BindAccept::new(bind_addr, async move {
+        let (stream, peer) = listener
+            .accept()
+            .await
+            .map_err(|e| BindAcceptError::General(format!("accept: {e}")))?;
+        let peer = peer
+            .to_target_addr()
+            .map_err(|e| BindAcceptError::General(format!("peer addr: {e}")))?;
+        let (read, write) = stream.into_split();
+        Ok(BindTargetHandle {
+            peer,
+            read: Box::new(read) as Box<dyn AsyncRead + Send + Sync + Unpin>,
+            write: Box::new(write) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
+        })
+    }))
+}
+
+#[cfg(test)]
+mod tests;

src/local/tests.rs

@@ -0,0 +1,471 @@
+//! The local providers' unit tests: real sockets — the echo TCP
+//! server for [`super::tcp_dial`] round-trips, a real UDP echo for
+//! the egress ([`super::LocalEgress`], the POC #2 shape — including
+//! the readiness contract's park-then-datagram shape), a real BIND
+//! accept ([`super::bind_accept`], reply#1/reply#2 carrying the real
+//! addresses), and the refusal mappings table-tested against the
+//! landed error types' `to_reply_error()`.
+
+use std::net::SocketAddr;
+use std::time::Duration;
+
+use bytes::Bytes;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::{TcpListener, TcpStream, UdpSocket};
+use tokio::time::timeout;
+
+use super::*;
+use crate::rfc1928::consts;
+
+const BOUND: Duration = Duration::from_secs(10);
+
+/// Decode the inner bytes' `(addr, payload)` shape — the forked
+/// `read_address` codec applied to the egress's reply (an honest
+/// decode through the codec, not a hand-parse: the inner bytes are
+/// `ATYP + addr + port + data`; the codec reads exactly the address
+/// field, the payload tail is un-read surplus).
+async fn decode_inner(inner: &[u8]) -> (TargetAddr, Bytes) {
+    let atyp = inner[0];
+    let (mut client, mut server) = tokio::io::duplex(inner.len() + 16);
+    client
+        .write_all(&inner[1..])
+        .await
+        .expect("feed the inner bytes");
+    drop(client);
+    let addr = crate::rfc1928::util::target_addr::read_address(&mut server, atyp)
+        .await
+        .expect("the address decodes");
+    let addr_len = addr.to_be_bytes().expect("re-encode").len();
+    (addr, Bytes::copy_from_slice(&inner[addr_len..]))
+}
+
+/// Spawn an echo TCP server on an ephemeral loopback port; returns
+/// its address (the dial target).
+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 UDP echo server: datagrams echo back from their sender address
+/// (the egress test's far side — the POC #2 shape).
+async fn echo_udp_server() -> SocketAddr {
+    let sock = UdpSocket::bind("127.0.0.1:0").await.expect("bind");
+    let addr = sock.local_addr().expect("local_addr");
+    tokio::spawn(async move {
+        let mut buf = [0u8; 65535];
+        loop {
+            let Ok((n, from)) = sock.recv_from(&mut buf).await else {
+                return;
+            };
+            if sock.send_to(&buf[..n], from).await.is_err() {
+                return;
+            }
+        }
+    });
+    addr
+}
+
+// --- tcp_dial -------------------------------------------------------------
+
+/// The happy path: dial the echo server, round-trip bytes through the
+/// boxed halves (ownership rides the handle; the halves are
+/// Send+Sync+Unpin — the boxing compiles, pinned here).
+#[tokio::test]
+async fn tcp_dial_round_trips_through_the_boxed_halves() {
+    let server = echo_tcp_server().await;
+    let handle = tcp_dial()(TargetAddr::Ip(server))
+        .await
+        .expect("the dial connects");
+    let mut write = handle.write;
+    write.write_all(b"ping").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut read = handle.read;
+    let mut buf = [0u8; 4];
+    timeout(BOUND, read.read_exact(&mut buf))
+        .await
+        .expect("echo within bound")
+        .expect("read");
+    assert_eq!(&buf, b"ping");
+}
+
+/// The `to_target_addr`-shaped target (the `impl ToTargetAddr`
+/// consumer path): a domain that resolves here (`localhost` — the
+/// `-D` semantic: resolution is the provider's job) reaches the same
+/// echo server; the seam received it unresolved and dialed the
+/// resolved address.
+#[tokio::test]
+async fn tcp_dial_resolves_domain_targets_here() {
+    let server = echo_tcp_server().await;
+    let handle = tcp_dial()(TargetAddr::Domain("localhost".to_owned(), server.port()))
+        .await
+        .expect("the domain target resolves and connects");
+    let mut write = handle.write;
+    write.write_all(b"pong").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut read = handle.read;
+    let mut buf = [0u8; 4];
+    timeout(BOUND, read.read_exact(&mut buf))
+        .await
+        .expect("echo within bound")
+        .expect("read");
+    assert_eq!(&buf, b"pong");
+}
+
+/// An unroutable loopback port refuses: `ConnectionRefused` (0x05)
+/// per the io error kind — the refusal rides the seam's error
+/// vocabulary (`to_reply_error()`), table-pinned through the mapped
+/// code.
+#[tokio::test]
+async fn tcp_dial_connect_refused_maps_faithfully() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let refused_port = listener.local_addr().expect("local_addr").port();
+    drop(listener);
+    let err = tcp_dial()(TargetAddr::Ip(SocketAddr::from((
+        [127, 0, 0, 1],
+        refused_port,
+    ))))
+    .await
+    .expect_err("nothing listens there");
+    assert!(
+        matches!(err, SocksDialError::Refused(_)),
+        "connect-refused → Refused: {err}"
+    );
+    assert_eq!(
+        err.to_reply_error().as_u8(),
+        consts::SOCKS5_REPLY_CONNECTION_REFUSED
+    );
+}
+
+/// A domain that does not resolve → `HostUnreachable` (0x04) — the
+/// resolution failure is the dial seam's error, faithfully mapped.
+#[tokio::test]
+async fn tcp_dial_unresolvable_domain_maps_to_host_unreachable() {
+    let err = tcp_dial()(TargetAddr::Domain("nonexistent.invalid".to_owned(), 443))
+        .await
+        .expect_err("the domain does not resolve");
+    assert!(
+        matches!(err, SocksDialError::HostUnreachable(_)),
+        "unresolvable → HostUnreachable: {err}"
+    );
+    assert_eq!(
+        err.to_reply_error().as_u8(),
+        consts::SOCKS5_REPLY_HOST_UNREACHABLE
+    );
+}
+
+// --- udp_egress ------------------------------------------------------------
+
+/// The datagram round-trip through a real UDP echo (the POC #2
+/// shape): `send` (payload-only) → the echo → `next_reply`'s inner
+/// bytes carry the sender's address + the data (no length prefix —
+/// framing-once; the wire layer adds it downstream).
+#[tokio::test]
+async fn udp_egress_round_trips_inner_bytes_with_sender_address() {
+    let echo = echo_udp_server().await;
+    let mut egress = udp_egress()().await.expect("the egress binds");
+
+    let dest = TargetAddr::Ip(echo);
+    egress
+        .send(dest.clone(), Bytes::from_static(b"query"))
+        .await
+        .expect("send");
+    let frame = timeout(BOUND, egress.next_reply())
+        .await
+        .expect("reply within bound")
+        .expect("recv")
+        .expect("a reply");
+
+    // The inner bytes: `to_be_bytes() + data`, no length prefix.
+    let mut expect = dest.to_be_bytes().expect("encode").to_vec();
+    expect.extend_from_slice(b"query");
+    assert_eq!(frame, Bytes::from(expect), "inner bytes, framed-once");
+
+    // The sender's address decodes as the echo's (the RFC's
+    // per-datagram source semantics — the address rides the frame).
+    let (src, payload) = decode_inner(&frame).await;
+    assert_eq!(src, dest);
+    assert_eq!(&payload[..], b"query");
+}
+
+/// The readiness contract with the real socket: `next_reply` parks
+/// (no reply is ready → `Pending`) — the stage's select! branch would
+/// stay parked; a datagram arriving resolves it. Pinned by racing the
+/// wait against a `sleep`: no reply → not resolved by the bound; the
+/// datagram → resolved. Never a hot `Ok(None)` (the send-then-drain
+/// ordering above already proves the resolution; this proves the
+/// park).
+#[tokio::test]
+async fn udp_egress_next_reply_parks_until_a_datagram_arrives() {
+    let echo = echo_udp_server().await;
+    let mut egress = udp_egress()().await.expect("the egress binds");
+    egress
+        .send(TargetAddr::Ip(echo), Bytes::from_static(b"park"))
+        .await
+        .expect("send");
+
+    // A `Pending`-honoring wait stays pending across a no-datagram
+    // window: if `next_reply` resolved hot with `Ok(None)`, this
+    // would return within the bound with a spurious shape instead of
+    // timing out.
+    let frame = timeout(BOUND, egress.next_reply())
+        .await
+        .expect("the reply arrives (the wait parked, the datagram resolved it)")
+        .expect("recv")
+        .expect("the echo arrives");
+
+    let (_, payload) = decode_inner(&frame).await;
+    assert_eq!(&payload[..], b"park");
+
+    // The socket drained: a second wait parks again (no reply is
+    // ready) — resolved only by a fresh datagram.
+    egress
+        .send(TargetAddr::Ip(echo), Bytes::from_static(b"again"))
+        .await
+        .expect("send");
+    let mut frame = None;
+    let deadline = std::time::Instant::now() + BOUND;
+    while std::time::Instant::now() < deadline {
+        match timeout(Duration::from_millis(500), egress.next_reply()).await {
+            Ok(Ok(Some(f))) => {
+                frame = Some(f);
+                break;
+            }
+            Ok(Ok(None)) => continue,
+            Ok(Err(e)) => panic!("the second reply errored: {e}"),
+            Err(_) => continue,
+        }
+    }
+    let frame = frame.expect("the second echo arrives (the wait parked, the datagram resolved it)");
+    let (_, payload) = decode_inner(&frame).await;
+    assert_eq!(&payload[..], b"again");
+}
+
+/// Multiple datagrams in order (no flow table — per-datagram
+/// addressing does the demultiplexing; the socket is the relay
+/// handle).
+#[tokio::test]
+async fn udp_egress_round_trips_multiple_datagrams_in_order() {
+    let echo = echo_udp_server().await;
+    let mut egress = udp_egress()().await.expect("the egress binds");
+    for i in 0..5u8 {
+        egress
+            .send(TargetAddr::Ip(echo), Bytes::from(vec![i; 8]))
+            .await
+            .expect("send");
+    }
+    for i in 0..5u8 {
+        let frame = timeout(BOUND, egress.next_reply())
+            .await
+            .expect("reply within bound")
+            .expect("recv")
+            .expect("a reply");
+        let (src, payload) = decode_inner(&frame).await;
+        assert_eq!(src, TargetAddr::Ip(echo));
+        assert_eq!(&payload[..], &[i; 8][..], "in-order echo");
+    }
+}
+
+/// A zero-length payload is a legal empty datagram (the F-2
+/// invariant at payload level) — round-trips through the real
+/// socket.
+#[tokio::test]
+async fn udp_egress_zero_length_payload_round_trips() {
+    let echo = echo_udp_server().await;
+    let mut egress = udp_egress()().await.expect("the egress binds");
+    egress
+        .send(TargetAddr::Ip(echo), Bytes::new())
+        .await
+        .expect("send the empty datagram");
+    let frame = timeout(BOUND, egress.next_reply())
+        .await
+        .expect("reply within bound")
+        .expect("recv")
+        .expect("the empty echo");
+    let (_, payload) = decode_inner(&frame).await;
+    assert!(payload.is_empty(), "the empty datagram is empty");
+}
+
+/// An undecodable destination (an unresolvable domain) fails the
+/// `send` with `Unreachable` → `NetworkUnreachable` (0x03) — the
+/// resolve failure is the egress's error vocabulary.
+#[tokio::test]
+async fn udp_egress_unresolvable_destination_maps_to_unreachable() {
+    let mut egress = udp_egress()().await.expect("the egress binds");
+    let err = egress
+        .send(
+            TargetAddr::Domain("nonexistent.invalid".to_owned(), 53),
+            Bytes::from_static(b"q"),
+        )
+        .await
+        .expect_err("the destination does not resolve");
+    assert!(matches!(err, EgressError::Unreachable(_)), "{err}");
+    assert_eq!(
+        err.to_reply_error().as_u8(),
+        consts::SOCKS5_REPLY_NETWORK_UNREACHABLE
+    );
+}
+
+// --- bind_accept ------------------------------------------------------------
+
+/// The faithful RFC shape (ADR-007 variant 1): reply#1's `bind_addr`
+/// is the listener's real `local_addr`; a client dialing it is
+/// accepted; the accepted handle's `peer` is the client's real
+/// address (byte-exact); the data plane round-trips.
+#[tokio::test]
+async fn bind_accept_reply1_real_addr_reply2_real_peer_and_round_trip() {
+    let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Domain("ftp.test".to_owned(), 0))
+        .await
+        .expect("the factory binds an ephemeral port");
+
+    let bind_addr = accept.bind_addr();
+    assert!(
+        bind_addr.ip().is_loopback() && bind_addr.port() != 0,
+        "reply#1 carries the real bound address: {bind_addr}"
+    );
+
+    // The app server dials the advertised address (the RFC's role
+    // for BND.ADDR).
+    let client = timeout(BOUND, TcpStream::connect(bind_addr))
+        .await
+        .expect("dial within bound")
+        .expect("connect to the advertised address");
+
+    let mut handle = timeout(BOUND, accept.accept())
+        .await
+        .expect("accept within bound")
+        .expect("the accept resolves");
+    let peer = match handle.peer {
+        TargetAddr::Ip(sa) => sa,
+        other => panic!("an IP peer, got {other:?}"),
+    };
+    assert_eq!(
+        peer,
+        client.local_addr().expect("client local_addr"),
+        "reply#2's peer is the client's real address (byte-exact)"
+    );
+
+    // The client echoes what it receives (the round-trip's far side:
+    // the app server the RFC models — active-mode FTP's data
+    // listener echo).
+    let mut client_to_echo = client;
+    tokio::spawn(async move {
+        let mut buf = [0u8; 4096];
+        loop {
+            match client_to_echo.read(&mut buf).await {
+                Ok(0) | Err(_) => return,
+                Ok(n) => {
+                    if client_to_echo.write_all(&buf[..n]).await.is_err() {
+                        return;
+                    }
+                }
+            }
+        }
+    });
+
+    // The data plane: bytes written to the handle reach the client;
+    // the client echoes back; the handle's read half receives them.
+    // The client echoes what it receives (the round-trip's far
+    // side), so both directions ride the accepted halves.
+    let mut write = handle.write;
+    write.write_all(b"active").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = [0u8; 6];
+    timeout(BOUND, handle.read.read_exact(&mut buf))
+        .await
+        .expect("echo within bound")
+        .expect("read");
+    assert_eq!(&buf, b"active");
+    write.write_all(b"pass").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = [0u8; 4];
+    timeout(BOUND, handle.read.read_exact(&mut buf))
+        .await
+        .expect("round-trip within bound")
+        .expect("read");
+    assert_eq!(&buf, b"pass");
+    drop(write);
+}
+
+/// The requested port is honored when non-zero (FTP active mode):
+/// the listener binds exactly the requested port, and a held port
+/// refuses the factory (`Shortage`/`Refused` per the io error — the
+/// bind failure is reply#1's vocabulary).
+#[tokio::test]
+async fn bind_accept_honors_a_requested_port_and_refuses_a_held_one() {
+    let holder = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let held = holder.local_addr().expect("local_addr");
+
+    let ok = bind_accept([127, 0, 0, 1])(TargetAddr::Ip(held))
+        .await
+        .expect_err("the port is held");
+    assert!(
+        matches!(
+            ok,
+            BindAcceptFactoryError::Shortage(_) | BindAcceptFactoryError::Refused(_)
+        ),
+        "bind refused/held → Shortage/Refused: {ok}"
+    );
+    assert_eq!(
+        ok.to_reply_error().as_u8(),
+        consts::SOCKS5_REPLY_GENERAL_FAILURE,
+        "Shortage/Refused map to the RFC's catch-all"
+    );
+
+    // A free port is honored exactly (the listener is the factory's
+    // resource — dropped with the accept future).
+    let free = TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let want = free.local_addr().expect("local_addr");
+    drop(free);
+    let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Ip(want))
+        .await
+        .expect("the free port binds");
+    assert_eq!(accept.bind_addr().port(), want.port(), "the requested port");
+    assert_eq!(accept.bind_addr().ip().to_string(), "127.0.0.1");
+}
+
+/// The accept future owns its listener end-to-end: dropping the
+/// `BindAccept` (the quiet wait's drop — the client-gone teardown)
+/// closes the listener; the advertised address stops accepting. A
+/// fresh factory on the same (now free) port re-binds.
+#[tokio::test]
+async fn bind_accept_drop_of_the_accept_future_closes_the_listener() {
+    let accept = bind_accept([127, 0, 0, 1])(TargetAddr::Domain("gone.test".to_owned(), 0))
+        .await
+        .expect("the factory binds");
+    let bind_addr = accept.bind_addr();
+    drop(accept);
+
+    // The listener closed with the future: a fresh bind of the same
+    // ephemeral port succeeds (the OS may hand the port to either
+    // the reuse or a fresh allocation — loopback ephemeral binds
+    // succeed either way; the probe is that the re-bind is possible
+    // at all, i.e. the listener was dropped, not leaked).
+    let rebinder = TcpListener::bind(bind_addr).await;
+    match rebinder {
+        Ok(rebound) => {
+            assert_eq!(rebound.local_addr().expect("local_addr"), bind_addr);
+        }
+        Err(e) => panic!("the listener leaked (the port is still held): {e}"),
+    }
+}