alksocks 0.2.0

//! The producer half: the open op spec (scope-gated), the
//! session-gate establisher, the command-dispatch pump handler
//! (typestate driving + the three seams), teardown.
//!
//! Layering (producer.md, the POC-validated shape): the establisher
//! is the **session gate** (`Establishment::default()` — the target
//! is unknown at open time; nothing to thread establisher→handler,
//! POC #1's wire note), the open handler is the **RFC state machine +
//! command dispatch** (the forked typestate, ADR-002), the I/O planes
//! are the **three seams** (dial / egress / accept, ADR-004).
//!
//! ## The establisher
//!
//! [`socks5_session_establisher`] validates/records the opener
//! identity (the CF-005/CF-006 seam — ADR-005: the producer side is
//! identity-only; authorization already happened at the open gate)
//! and applies session-level policy. It returns
//! `Establishment::default()` — no plan payload: the dial happens
//! post-establishment, in-stream at command dispatch, so there is
//! nothing to thread establisher→handler. The optional
//! `identity_witness` probe seam (the alktunnels posture) records
//! the per-call opener identity for tests.
//!
//! A channels-level refusal is the registry ACL gate's job (it runs
//! before the establisher — `socks5:open`, ADR-006; the scope-less
//! identity fails closed with `FORBIDDEN`, no phantom channel). The
//! establisher itself never refuses — it is the identity-recording
//! session gate (ADR-005).
//!
//! ## The pump handler + dispatch
//!
//! [`make_socks5_pump_handler`] accepts the channel's `BiStream`
//! (`accept_bi` — the yield-once contract) and drives the forked
//! typestate over it (`accept_no_auth → read_command`, ADR-005), then
//! dispatches:
//!
//! - **CONNECT** → the dial seam ([`DialFn`], ADR-004) at the
//!   command-read interception point; success: `reply_success` then
//!   `pump_bidi` **awaited inline** (R-02 — the returned `JoinHandle`
//!   tracks the data-plane lifetime; early return = teardown-at-
//!   birth). Refusal: `reply_error` with the mapped RFC code
//!   (dial-failure → `ConnectionRefused` 0x05; scope-class →
//!   `ConnectionNotAllowed` 0x02) — no phantom data plane.
//! - **UDP ASSOCIATE** → the datagram stage (ADR-003): the sentinel
//!   reply (`127.0.0.1:0` — the relay handle IS the channel; the
//!   address is pure RFC-compatibility surface), then the
//!   `select!`-driven stage (wire.md §UDP ASSOCIATE): channel reads ∥
//!   egress replies, framing applied exactly once by the wire layer,
//!   the egress readiness contract (`Pending` when nothing is ready),
//!   association lifetime = channel lifetime. Missing egress seam:
//!   RFC-faithful refusal (`CommandNotSupported` — ADR-008's
//!   missing-callback class). Egress-factory failure: reply#1 with
//!   the mapped code.
//! - **TCP BIND** → the command gate first (`socks5:bind`, ADR-006,
//!   from the per-call opener identity — a scope-less identity gets
//!   `ConnectionNotAllowed` 0x02 while CONNECT still serves it), then
//!   the phase model (ADR-007): the accept seam
//!   ([`BindAcceptFactory`], ADR-004) at the command-read
//!   interception point, reply#1 carrying BND.ADDR, the quiet wait
//!   (single-shot, exactly two exits — the accept, or the control
//!   stream's EOF/garbage/error), reply#2 carrying the peer address,
//!   then `pump_bidi` inline. Two-stage failure mapping: factory
//!   failure → reply#1's code; accept-stage failure → reply#2's
//!   code, no data plane. No factory wired = the wrapper-aware
//!   in-band default (ADR-007 variant 3): sentinel reply#1 (via the
//!   accept future's instant resolution) + sentinel reply#2
//!   (`0.0.0.0:0`).
//!
//! The handler is substrate-agnostic by construction — it sees boxed
//! halves, never a socket type (AGENTS.md convention 7; the
//! inversion point is the seam closure, ADR-004).
//!
//! ## `serve_socks5_over<T>`
//!
//! [`serve_socks5_over`] is the substrate-free protocol entry point
//! (the POC shape): generic over `T: AsyncRead + AsyncWrite + Unpin`,
//! driving the typestate and dispatching, with the `can_bind` gate
//! carried explicitly (the open-op identity seam is consumed by the
//! pump handler — the substrate-free entry takes the resolved gate
//! boolean). The channel `BiStream` and a `TcpStream` are equally
//! `T` — this is what makes the producer testable without channels
//! (the differential-suite pattern) and reusable by the batch-2
//! local backend (feature `local`, ADR-009). The builder's shape
//! resolves that gate from its explicit `bind_policy` (defaulting to
//! the accept seam's presence — never the sole gate; ADR-006's
//! "listener configured" ≠ "listener permitted").
//!
//! ## Registration
//!
//! [`register_socks5_openable`] wires spec + establisher + pump via
//! `ChannelCore::register_openable_with_establisher` (`timeout: None`
//! = the 10s default bound — ADR-049 §2), scope-gated `socks5:open`
//! (the registry runs the ACL before the establisher). Missing
//! callbacks = those commands refuse RFC-faithfully
//! (`CommandNotSupported`-class), never a panic (ADR-008). The
//! builder ([`Socks5Producer::builder`]) is the spec'd assembly
//! shape (producer.md §Wiring); the post-hoc per-connection shape
//! (the harness's W2) rides the same registration.

use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;

use alkcall::channels::operations::{ChannelCore, Establishment, OpenEstablisher, OpenHandler};
use alkcall::channels::pump::pump_bidi;
use alkcall::core::auth::AuthContext;
use alkcall::core::types::Connection;
use alkcall::registry::registration::OperationRegistry;
use serde_json::Value;
use tokio::io::{AsyncRead, AsyncWrite};

use crate::egress::{
    BindAccept, BindAcceptFactory, BindTargetHandle, DialFn, Egress, EgressFactory,
};
use crate::params::{socks5_open_spec, SOCKS5_BIND_SCOPE};
use crate::rfc1928::server::{new_reply, Socks5ServerProtocol, SocksServerError};
use crate::rfc1928::util::target_addr::TargetAddr;
use crate::rfc1928::{ReplyError, Socks5Command, SocksError};
use crate::wire::{frame_inner, DatagramReader};

pub use crate::params::{OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_OPEN_SCOPE};

/// The sentinel `BND.ADDR` the in-band reply carries (ADR-003): pure
/// RFC-compatibility surface — the relay handle IS the channel;
/// wrapper-aware clients never sendto() it (the datagram stage is
/// in-band) and never bind() it (the accept variants own the inbound
/// surface). `127.0.0.1:0` per wire.md §UDP ASSOCIATE.
pub(crate) fn sentinel_reply_addr() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0)
}

/// What the establisher's per-call auth carried (the CF-006 probe):
/// the opener identity id, `None` until an open runs. The optional
/// `identity_witness` seam (the alktunnels posture) for tests.
pub type IdentityWitness = Arc<tokio::sync::Mutex<Option<String>>>;

/// The session-gate establisher: the open op's awaited phase. For the
/// SOCKS5 op it records the opener identity (the CF-005/CF-006 seam
/// — ADR-005; per-call `auth` is the dispatch-resolved identity
/// overlaid onto the install-time context) and applies session-level
/// policy. A channels-level refusal is the registry ACL gate's job —
/// it runs before this ever runs (ADR-006). The RFC handshake is
/// in-stream, after this. Returns `Establishment::default()` — no
/// plan needed (the dial is in-stream, in the pump handler; POC #1's
/// wire note).
pub fn socks5_session_establisher(witness: Option<IdentityWitness>) -> OpenEstablisher {
    Arc::new(move |_input: Value, auth: AuthContext| {
        let witness = witness.clone();
        Box::pin(async move {
            if let Some(witness) = witness {
                *witness.lock().await = auth.identity.as_ref().map(|i| i.id.clone());
            }
            Ok(Establishment::default())
        })
    })
}

/// The pump handler: accept the channel's yield-once `BiStream`
/// (`accept_bi` first — the wrapper's accepted-flag telemetry), then
/// run the RFC 1928 state machine over it and dispatch on the
/// command. The command gate rides the open-op `AuthContext`'s
/// per-call opener identity (CF-005/CF-006): TCP BIND requires
/// `socks5:bind` (ADR-006) — CONNECT and UDP ASSOCIATE need only the
/// open gate. The handler is substrate-agnostic by construction — it
/// sees boxed halves, never a socket type (ADR-004's inversion
/// point). Awaits the data plane inline (R-02 — the returned
/// `JoinHandle` tracks the data-plane lifetime; early return =
/// teardown-at-birth).
pub fn make_socks5_pump_handler(
    dial: Option<DialFn>,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
) -> OpenHandler {
    Arc::new(
        move |_input: Value,
              _plan: Option<alkcall::channels::operations::ChannelPlan>,
              conn: Connection,
              auth: AuthContext| {
            let dial = dial.clone();
            let egress = egress.clone();
            let accept = accept.clone();
            let can_bind = auth
                .identity
                .as_ref()
                .map(|i| i.scopes.iter().any(|s| s == SOCKS5_BIND_SCOPE))
                .unwrap_or(false);
            tokio::spawn(async move {
                let Ok(bidi) = conn.accept_bi().await else {
                    tracing::error!("socks5 pump: accept_bi failed");
                    return;
                };
                if let Err(err) = serve_socks5_over(bidi, dial, egress, accept, can_bind).await {
                    tracing::warn!("socks5 session ended: {err:#}");
                }
            })
        },
    )
}

/// The RFC 1928 server conversation over any `T` — the substrate-free
/// protocol layer (the POC shape). `T` is the channel `BiStream`, a
/// `TcpStream`, a duplex half — the typestate and the seams never
/// know. `dial`/`egress`/`accept` are the injected seams (ADR-004);
/// a missing seam makes its command refuse RFC-faithfully
/// (`CommandNotSupported`-class, ADR-008) — never a panic. `can_bind`
/// is the command gate's resolved boolean (ADR-006 — the pump handler
/// resolves it from the per-call opener identity; a substrate-free
/// caller passes its policy's answer).
pub async fn serve_socks5_over<T>(
    stream: T,
    dial: Option<DialFn>,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    can_bind: bool,
) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    let (proto, cmd, target) = Socks5ServerProtocol::accept_no_auth(stream)
        .await?
        .read_command()
        .await?;

    match cmd {
        Socks5Command::TCPConnect => serve_connect(proto, target, dial).await,
        Socks5Command::UDPAssociate => serve_udp_associate(proto, egress).await,
        Socks5Command::TCPBind => {
            if !can_bind {
                // The command gate (ADR-006): BIND mints inbound
                // listeners — an identity with proxy egress must not
                // imply ingress. RFC-faithful refusal (the command is
                // known, the caller is not allowed). The session
                // error is the log/observability vocabulary (the wire
                // got its reply already) — `Refused`, the honest
                // shape (review 002 N-4; an ACL refusal is not an
                // auth failure).
                proto.reply_error(&ReplyError::ConnectionNotAllowed).await?;
                return Err(SocksError::ServerError(SocksServerError::Refused(
                    "bind command gate: not allowed",
                )));
            }
            serve_tcp_bind(proto, target, accept).await
        }
    }
}

/// CONNECT: dial at the command-read interception point (ADR-004),
/// reply, then the pass-through data plane (`pump_bidi`, ADR-050's
/// two-pump shape, awaited inline — R-02). Refusal: `reply_error`
/// with the mapped RFC code — no phantom data plane.
async fn serve_connect<T>(
    proto: Socks5ServerProtocol<T, crate::rfc1928::server::states::CommandRead>,
    target: TargetAddr,
    dial: Option<DialFn>,
) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    let Some(dial) = dial else {
        proto.reply_error(&ReplyError::CommandNotSupported).await?;
        return Err(SocksError::ServerError(SocksServerError::UnknownCommand(
            Socks5Command::TCPConnect.as_u8(),
        )));
    };

    let handle = match dial(target).await {
        Ok(handle) => handle,
        Err(err) => {
            proto.reply_error(&err.to_reply_error()).await?;
            return Err(SocksError::ServerError(SocksServerError::Refused(
                "dial refused in-stream",
            )));
        }
    };

    let inner = proto.reply_success(sentinel_reply_addr()).await?;

    pump_bidi(inner, handle.read, handle.write).await;
    Ok(())
}

/// UDP ASSOCIATE: the datagram stage (ADR-003, POC #2-validated).
/// The typestate is driven directly (no `run_udp_proxy_custom` — its
/// unconditional kernel bind is unusable on channels); the reply
/// carries the sentinel address (the relay handle IS the channel;
/// the address is pure RFC-compatibility surface for wrapper-aware
/// clients, who never sendto() it); the post-reply stream is the
/// framed `{addr, data}` datagram stage. A missing egress seam
/// refuses RFC-faithfully (ADR-008's missing-callback class — an
/// unprovided capability, never a panic); an egress-factory failure
/// refuses with the mapped code.
async fn serve_udp_associate<T>(
    proto: Socks5ServerProtocol<T, crate::rfc1928::server::states::CommandRead>,
    egress: Option<EgressFactory>,
) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    let Some(egress) = egress else {
        proto.reply_error(&ReplyError::CommandNotSupported).await?;
        return Err(SocksError::ServerError(SocksServerError::UnknownCommand(
            Socks5Command::UDPAssociate.as_u8(),
        )));
    };

    let mut egress = match egress().await {
        Ok(egress) => egress,
        Err(err) => {
            proto.reply_error(&err.to_reply_error()).await?;
            return Err(SocksError::ServerError(SocksServerError::Refused(
                "egress refused in-stream",
            )));
        }
    };

    let inner = proto.reply_success(sentinel_reply_addr()).await?;

    run_datagram_stage(inner, egress.as_mut()).await;
    Ok(())
}

/// The post-reply datagram stage (wire.md §UDP ASSOCIATE; the POC #2
/// shape). Three POC invariants, load-bearing:
///
/// - **The concurrent select shape** — channel reads and the egress's
///   reply stream run concurrently (the honest full-duplex shape; a
///   poll-then-block loop loses replies arriving while the channel
///   read is parked — the POC's first iteration, measured).
/// - **The egress readiness contract** (`Egress::next_reply`): it
///   must be `Pending` when no reply is ready; an instantly-ready
///   `Ok(None)` starves a biased select into livelock. The stage
///   tolerates a spurious `Ok(None)` by yielding once and re-parking
///   (never a hot loop — and never a zero-yield hot spin on a
///   current-thread scheduler, the livelock a pure `async { Ok(None)
///   }` egress reproduces); an `Err` ends the association (the egress
///   drops with it).
/// - **Framing applied exactly once, by the wire layer** — the stage
///   reads frames via `DatagramReader` (channel → egress `send` gets
///   the payload) and writes frames via `frame_inner` (egress
///   `next_reply`'s inner bytes → the channel). The egress never sees
///   a length prefix.
///
/// Association lifetime = channel lifetime: channel EOF ends the
/// stage; no extra control vocabulary. A undecodable frame (garbage
/// ATYP — any post-reply control byte) fails the association loud
/// (the RFC's "post-reply control data is garbage" semantic). A
/// stream end mid-frame is fail-loud too (never a silent partial
/// datagram — the reader state is the teardown diagnostic). The
/// stage owns the egress: on every exit path the egress binding
/// drops here (teardown-at-birth — no `_` bindings of owning
/// values).
///
/// Per-datagram addressing in the frame does the demultiplexing (no
/// flow table); one relay loop, no spawn-per-datagram. A frame's
/// egress `send` is awaited inline before the next select iteration —
/// the stage interleaves egress sends with reply drains (the POC's
/// measured shape; sufficient for relay postures; a slow egress
/// applies backpressure to the association, never to other channels —
/// wire.md §UDP ASSOCIATE).
///
/// Residual posture (review 002 S-1, named decision): the `Ok(None)`
/// tolerance's defense is bounded by `yield_now` only — a *misbehaving*
/// egress returning `Ok(None)` at full speed degrades the stage into
/// a yielding busy-wait (making no progress, unbounded by rate; the
/// runtime survives, the stage never parks). A compliant egress parks
/// (the readiness contract); a backoff/budget in the stage would
/// penalize the compliant path, so the residual is bounded by the
/// egress's own correctness and accepted-uncovered — named here, not
/// accidental.
async fn run_datagram_stage<T>(mut inner: T, egress: &mut (dyn Egress + Send)) -> StageExit
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut reader = DatagramReader::new();
    let mut chunk = vec![0u8; 16 * 1024];
    loop {
        tokio::select! {
            // The egress's reply stream (its own pump; `None` = a
            // spurious wakeup — the stage re-parks, never a hot
            // loop). The yield makes the tolerance safe against a
            // contract-violating zero-yield egress too: without it a
            // pure `async { Ok(None) }` starves the channel branch
            // and the runtime's other tasks on a current-thread
            // scheduler (an instantly-ready branch that yields
            // nothing wins every poll — the readiness contract's POC
            // #2 finding, hardened by the gate review's R-1
            // (tasks/implementation/review-protocol-core.md)).
            reply = egress.next_reply() => {
                match reply {
                    Ok(Some(frame)) => {
                        match frame_inner(&frame) {
                            Ok(wire) => {
                                if inner.write_all(&wire).await.is_err() {
                                    return StageExit::Io;
                                }
                            }
                            Err(err) => {
                                tracing::debug!("udp stage: reply frame rejected: {err}");
                                return StageExit::ReplyFrameRejected;
                            }
                        }
                    }
                    Ok(None) => tokio::task::yield_now().await,
                    Err(err) => {
                        tracing::debug!("udp egress relay ended: {err:#}");
                        return StageExit::EgressEnded;
                    }
                }
            }
            // The channel's read side: frames in, EOF = end.
            read = inner.read(&mut chunk) => {
                match read {
                    // Channel EOF = association end (the clean exit —
                    // the association lifetime is the channel's).
                    Ok(0) => return StageExit::ChannelEof,
                    Ok(n) => {
                        for dg in reader.feed(&chunk[..n]) {
                            match dg {
                                Err(err) => {
                                    tracing::debug!("udp stage: undecodable frame: {err}");
                                    return StageExit::UndecodableFrame;
                                }
                                Ok((addr, payload)) => {
                                    if let Err(err) = egress.send(addr, payload).await {
                                        tracing::debug!("udp egress send failed: {err:#}");
                                        return StageExit::EgressEnded;
                                    }
                                }
                            }
                        }
                    }
                    Err(err) => {
                        tracing::debug!("udp stage: channel read error: {err}");
                        return StageExit::Io;
                    }
                }
            }
        }
    }
}

/// Why the datagram stage ended (the stage's teardown diagnostic —
/// each exit drops the egress with the stage; the channel EOF is the
/// one clean exit, per the association-lifetime contract).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum StageExit {
    /// Channel EOF — the association's clean end (the one the
    /// association-lifetime contract names).
    ChannelEof,
    /// A frame the codec refused (garbage ATYP / a `len == 0` frame —
    /// post-reply control bytes). Loud fail: never a silent drop.
    UndecodableFrame,
    /// An egress `next_reply` `Err` or a `send` failure (the egress
    /// relay ended; the lossy-datagram posture).
    EgressEnded,
    /// A reply the wire codec rejected (the egress violated the
    /// joint bound — never a wire overflow).
    ReplyFrameRejected,
    /// The channel stream itself failed (I/O).
    Io,
}

/// TCP BIND: the phase model (ADR-007; POC #5-validated). Phases:
///
/// 1. **Accept factory** — the CONNECT dial's mirror, called at the
///    command-read interception point with the requested target. A
///    factory failure maps to reply#1 (no phantom listener). No
///    factory wired = the wrapper-aware in-band default (ADR-007
///    variant 3): a synthetic accept whose reply#1 is the sentinel —
///    the handle-swap reading applied to BIND (the reply is
///    RFC-compatibility surface; the data plane is the channel).
/// 2. **Reply#1** — success + BND.ADDR (the factory's `bind_addr`:
///    the real bound listener on the provider-backed variants, the
///    sentinel on the wrapper-aware one).
/// 3. **The quiet wait** — single-shot `select!` with exactly two
///    exits (the POC's `never_loop` finding): the one inbound accept
///    (the `BindAccept` future — its drop IS the client-gone
///    teardown of whatever the provider opened), or the control
///    stream's EOF/garbage/error. Not a loop: any byte in the quiet
///    wait is protocol garbage (a BIND client never sends another
///    control byte — RFC 1928 §3's post-reply semantic) → the
///    session ends with no data plane. The wait lives in the
///    teardown-governed pump phase, never under the establishment
///    bound (FTP-length BIND waits stay legal).
/// 4. **Reply#2** — the peer address (the accepted handle's peer;
///    the wrapper-aware sentinel is `0.0.0.0:0`). An accept-stage
///    failure maps to reply#2's code — no data plane.
/// 5. **Data plane** — `pump_bidi` inline (the CONNECT shape, R-02):
///    raw pass-through between the channel and the accepted halves.
async fn serve_tcp_bind<T>(
    proto: Socks5ServerProtocol<T, crate::rfc1928::server::states::CommandRead>,
    target: TargetAddr,
    accept: Option<BindAcceptFactory>,
) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    let accept = match accept {
        Some(factory) => match factory(target).await {
            Ok(accept) => accept,
            Err(err) => {
                proto.reply_error(&err.to_reply_error()).await?;
                return Err(SocksError::ServerError(SocksServerError::Refused(
                    "bind factory refused in-stream",
                )));
            }
        },
        None => in_band_bind_accept(),
    };

    // Reply#1: success + BND.ADDR — the factory's address is the
    // carrier (real listener / advertised composed address / the
    // wrapper-aware sentinel).
    let inner = proto.reply_success(accept.bind_addr()).await?;

    // The quiet wait → reply#2 + the data plane.
    bind_quiet_wait_and_serve(inner, accept).await
}

/// The no-factory in-band default (ADR-007 variant 3, wrapper-aware):
/// the sentinel reply#1 rides a synthetic accept whose future
/// resolves instantly — the phase model is preserved end to end
/// (reply#1 → quiet wait → reply#2 → pass-through) with the channel
/// itself as the accepted "connection". Reply#2's peer is the
/// wrapper-aware sentinel `0.0.0.0:0`.
fn in_band_bind_accept() -> BindAccept {
    BindAccept::new(sentinel_bind_peer_addr(), async {
        Ok(BindTargetHandle {
            peer: TargetAddr::Ip(sentinel_bind_peer_addr()),
            read: Box::new(tokio::io::empty()) as Box<dyn AsyncRead + Send + Sync + Unpin>,
            write: Box::new(tokio::io::sink()) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
        })
    })
}

/// The quiet wait + reply#2 + the data plane. Single-shot `select!`
/// with exactly two exits (the accept, or the control stream's
/// EOF/garbage/error); the accept branch's error maps to reply#2's
/// code — no data plane; the accept success writes reply#2 with the
/// peer address then `pump_bidi`s inline.
async fn bind_quiet_wait_and_serve<T>(mut inner: T, accept: BindAccept) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    let mut chunk = [0u8; 1];
    let wait = accept.accept();
    tokio::pin!(wait);
    let handle = tokio::select! {
        handle = &mut wait => match handle {
            Ok(h) => h,
            Err(err) => {
                // Accept-stage failure (post-reply#1): reply#2
                // carries the mapped code; the session ends — no
                // data plane.
                let _ = inner
                    .write_all(&new_reply(&err.to_reply_error(), sentinel_bind_peer_addr()))
                    .await;
                let _ = inner.flush().await;
                return Err(SocksError::ServerError(SocksServerError::Refused(
                    "bind accept refused in-stream",
                )));
            }
        },
        read = inner.read(&mut chunk) => match read {
            // Client gone: the accept future drops here — whatever
            // the provider opened (the listener, the tunnel) is
            // owned by that future, so it tears down with the
            // session (the client-gone contract, ADR-004/ADR-007).
            Ok(0) => return Ok(()),
            // Any byte in the quiet wait is protocol garbage (a BIND
            // client never sends another control byte) — the session
            // ends with no data plane, no reply#2.
            Ok(_) => {
                return Err(SocksError::ServerError(
                    SocksServerError::UnexpectedUdpControlGarbage(chunk[0]),
                ));
            }
            Err(_) => return Ok(()),
        },
    };

    // Reply#2: success + the accepted peer address, then the raw
    // pass-through data plane (the CONNECT shape), awaited inline.
    let peer = match handle.peer {
        TargetAddr::Ip(addr) => addr,
        TargetAddr::Domain(_, port) => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port),
    };
    let mut inner = inner;
    inner
        .write_all(&new_reply(&ReplyError::Succeeded, peer))
        .await?;
    inner.flush().await?;

    pump_bidi(inner, handle.read, handle.write).await;
    Ok(())
}

/// The wrapper-aware sentinel peer (reply#2's BND.ADDR on the
/// in-band path): `0.0.0.0:0` per wire.md §BIND.
fn sentinel_bind_peer_addr() -> SocketAddr {
    SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0)
}

/// Register the SOCKS5 open op on a `ChannelCore` + the session's
/// dispatch registry: spec + establisher + pump + injected seams,
/// via `ChannelCore::register_openable_with_establisher` (`timeout:
/// None` = the 10s default bound, ADR-049 §2). Scope-gated
/// `socks5:open` (the registry runs the ACL before the establisher —
/// ADR-006). Missing callbacks = those commands refuse
/// RFC-faithfully (`CommandNotSupported`-class), never a panic
/// (ADR-008).
pub fn register_socks5_openable(
    core: &ChannelCore,
    on_registry: &Arc<OperationRegistry>,
    auth: AuthContext,
    dial: Option<DialFn>,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    witness: Option<IdentityWitness>,
) -> Result<(), String> {
    core.register_openable_with_establisher(
        socks5_open_spec(),
        Some(socks5_session_establisher(witness)),
        make_socks5_pump_handler(dial, egress, accept),
        on_registry,
        auth,
        None,
    )
}

/// The producer assembly shape (producer.md §Wiring): the
/// registration's injected seams as a builder — the assembly layer
/// closes over substrate code behind the callbacks; the protocol
/// layer never sees a socket type (ADR-004/ADR-008). Missing
/// callbacks = those commands refuse RFC-faithfully (ADR-008).
///
/// The BIND command gate on the substrate-free entry
/// ([`Socks5Producer::serve_socks5_over`]) is the builder's
/// `bind_policy` — explicit via `.bind_policy(bool)`, defaulting to
/// the accept seam's presence for backward-compat. The channels
/// path's gate is the pump handler's, resolved from the per-call
/// opener identity (ADR-006) — this policy never touches it.
#[derive(Clone, Default)]
pub struct Socks5Producer {
    dial: Option<DialFn>,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    bind_policy: bool,
}

impl Socks5Producer {
    pub fn builder() -> Socks5ProducerBuilder {
        Socks5ProducerBuilder::default()
    }

    /// The substrate-free protocol entry (the same shape
    /// [`serve_socks5_over`] pins) over the builder's seams. The
    /// command gate is the builder's `bind_policy` — an explicit
    /// `.bind_policy(bool)` whenever the local policy must speak
    /// ("factory wired but gate closed" is expressible here);
    /// unwired, it defaults to the accept seam's presence for
    /// backward-compat. The per-caller gate is the pump handler's,
    /// resolved from the open-op identity — ADR-006.
    pub async fn serve_socks5_over<T>(&self, stream: T) -> Result<(), SocksError>
    where
        T: AsyncRead + AsyncWrite + Unpin,
    {
        serve_socks5_over(
            stream,
            self.dial.clone(),
            self.egress.clone(),
            self.accept.clone(),
            self.bind_policy,
        )
        .await
    }

    /// Register the SOCKS5 open op — the same shape
    /// [`register_socks5_openable`] pins, over the builder's seams +
    /// the optional identity witness.
    pub fn register(
        &self,
        core: &ChannelCore,
        on_registry: &Arc<OperationRegistry>,
        auth: AuthContext,
        witness: Option<IdentityWitness>,
    ) -> Result<(), String> {
        register_socks5_openable(
            core,
            on_registry,
            auth,
            self.dial.clone(),
            self.egress.clone(),
            self.accept.clone(),
            witness,
        )
    }
}

/// The builder for [`Socks5Producer`].
#[derive(Default)]
pub struct Socks5ProducerBuilder {
    dial: Option<DialFn>,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    bind_policy: Option<bool>,
}

impl Socks5ProducerBuilder {
    /// The dial seam (required for CONNECT; ADR-004).
    pub fn dial(mut self, dial: DialFn) -> Self {
        self.dial = Some(dial);
        self
    }

    /// The egress seam (required for ASSOCIATE; ADR-004).
    pub fn egress(mut self, egress: EgressFactory) -> Self {
        self.egress = Some(egress);
        self
    }

    /// The accept seam (required for BIND; ADR-004).
    pub fn accept(mut self, accept: BindAcceptFactory) -> Self {
        self.accept = Some(accept);
        self
    }

    /// The substrate-free entry's BIND command gate (ADR-006's
    /// posture — "listener configured" ≠ "listener permitted"): the
    /// resolved `can_bind` for [`Socks5Producer::serve_socks5_over`].
    /// Set it when the local policy must close the gate while the
    /// accept seam stays wired. Unwired, the gate defaults to the
    /// accept seam's presence (the backward-compat reading — the
    /// pump handler's identity-derived gate is unaffected).
    pub fn bind_policy(mut self, can_bind: bool) -> Self {
        self.bind_policy = Some(can_bind);
        self
    }

    pub fn build(self) -> Socks5Producer {
        Socks5Producer {
            bind_policy: self.bind_policy.unwrap_or(self.accept.is_some()),
            dial: self.dial,
            egress: self.egress,
            accept: self.accept,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_common as common;

    use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError};
    use bytes::Bytes;
    use common::{
        datagram_frame, decode_datagram, echo_dial_witnessed, read_reply_v4, refusing_dial,
        write_request,
    };
    use std::pin::Pin;
    use std::task::{Context, Poll};
    use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};

    /// The test dial closure (the common module's witnessed echo
    /// shape — the target's identity is returned to the test through
    /// the shared witness so the interception-point shape is pinned).
    fn echo_dial(targets: Arc<tokio::sync::Mutex<Vec<TargetAddr>>>) -> DialFn {
        echo_dial_witnessed(targets)
    }

    /// Drive a raw RFC 1928 client against [`serve_socks5_over`] over
    /// a duplex (the differential-suite pattern: the producer
    /// testable without channels). `can_bind` defaults to the
    /// CONNECT-only shape (the gate's no-grant answer).
    async fn raw_connect(
        dial: DialFn,
    ) -> (
        tokio::task::JoinHandle<Result<(), SocksError>>,
        DuplexStream,
    ) {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None, false));
        (handle, client_end)
    }

    /// The same shape with the seams + gate chosen explicitly (the
    /// BIND tests' entry).
    async fn raw_connect_with(
        dial: Option<DialFn>,
        accept: Option<BindAcceptFactory>,
        can_bind: bool,
    ) -> (
        tokio::task::JoinHandle<Result<(), SocksError>>,
        DuplexStream,
    ) {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(server_end, dial, None, accept, can_bind));
        (handle, client_end)
    }

    async fn greet(client: &mut DuplexStream) {
        common::greet(client).await
    }

    async fn read_reply(client: &mut DuplexStream) -> (u8, [u8; 6]) {
        read_reply_v4(client).await
    }

    /// The full CONNECT conversation: dial is invoked at the
    /// command-read interception point, success reply, then the
    /// two-pump echo round-trip through the dialed target.
    #[tokio::test]
    async fn connect_round_trip_through_pump_bidi() {
        let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let (handle, mut client) = raw_connect(echo_dial(Arc::clone(&targets))).await;

        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("echo.test".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "the dial resolved");

        // The data plane is raw pass-through: echo through the dialed
        // target, both directions.
        client.write_all(b"ping").await.unwrap();
        let mut buf = [0u8; 4];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"ping");

        // The dial seam saw the parsed RFC target (the interception
        // point — domain form passes through unresolved, ADR-012).
        assert_eq!(
            targets.lock().await.last(),
            Some(&TargetAddr::Domain("echo.test".into(), 443))
        );

        // Half-close teardown: client EOF ends the session cleanly.
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends on client EOF")
            .expect("join");
        result.expect("clean session");
    }

    /// A dial refusal surfaces as the mapped RFC reply code in the
    /// reply — no phantom data plane (nothing is dialed, no bytes
    /// flow after the refusal).
    #[tokio::test]
    async fn dial_refusal_maps_to_reply_code_with_no_data_plane() {
        let dial = refusing_dial(SocksDialError::Refused("target down".into()));
        let (handle, mut client) = raw_connect(dial).await;

        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("down.test".into(), 80),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x05, "dial refusal → ConnectionRefused");

        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the session surfaces the refusal in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("dial refused in-stream"))
        ));
    }

    /// Every dial error variant maps to its documented RFC reply
    /// code through the full conversation (the in-stream refusal
    /// path, not just the mapping table).
    #[tokio::test]
    async fn dial_error_variants_map_to_their_reply_codes() {
        for (err, code) in [
            (SocksDialError::Refused("x".into()), 0x05),
            (SocksDialError::NetworkUnreachable("x".into()), 0x03),
            (SocksDialError::HostUnreachable("x".into()), 0x04),
            (SocksDialError::NotAllowed("x".into()), 0x02),
            (SocksDialError::General("x".into()), 0x01),
        ] {
            let (handle, mut client) = raw_connect(refusing_dial(err)).await;
            greet(&mut client).await;
            write_request(
                &mut client,
                &Socks5Command::TCPConnect,
                &TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 80))),
            )
            .await;
            let (got, _addr) = read_reply(&mut client).await;
            assert_eq!(got, code, "the mapped code rides the reply");
            drop(client);
            let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
                .await
                .expect("session ends")
                .expect("join");
        }
    }

    /// A missing dial callback refuses CONNECT RFC-faithfully
    /// (`CommandNotSupported` — ADR-008's posture: an unprovided
    /// capability, never a panic).
    #[tokio::test]
    async fn missing_dial_refuses_connect_command_not_supported() {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(server_end, None, None, None, false));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("no.dial".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x07, "missing dial → CommandNotSupported");
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal is surfaced in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::UnknownCommand(_))
        ));
    }

    // -- UDP ASSOCIATE: the datagram stage (ADR-003, POC #2 shape,
    //    in-process over duplexes) ------------------------------------
    // (the queue egress + its echo factory + the frame helpers ride
    // the common module — the harness's single-sourced shapes.)

    use crate::egress::EgressError;

    /// The full ASSOCIATE conversation over a duplex: greet →
    /// associate request → sentinel reply. Returns (server handle,
    /// client duplex). `egress_factory_err` simulates a refusing
    /// factory; `None` wires the echo egress.
    async fn raw_associate(
        egress: Option<EgressFactory>,
    ) -> (
        tokio::task::JoinHandle<Result<(), SocksError>>,
        tokio::io::DuplexStream,
    ) {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
        let handle = tokio::spawn(serve_socks5_over(
            server_end,
            Some(dial),
            egress,
            None,
            false,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("relay.test".into(), 1080),
        )
        .await;
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "the associate reply succeeds");
        assert_eq!(
            (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
            (127, 0, 0, 1, 0, 0),
            "the reply carries the sentinel BND.ADDR 127.0.0.1:0"
        );
        (handle, client)
    }

    /// The echo egress over the queue shape (the common module's
    /// `echo_egress_factory` — the harness's `echo_egress`, same
    /// shape): datagrams sent echo back as inner bytes.
    fn echo_egress_factory() -> EgressFactory {
        common::echo_egress_factory()
    }

    /// The ASSOCIATE happy path: sentinel reply → datagram
    /// round-trip through the echo egress (the frame shape both
    /// directions, in-order across multiple datagrams on one
    /// association).
    #[tokio::test]
    async fn udp_associate_round_trip_through_the_echo_egress() {
        let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;

        // Multiple datagrams, in order, one association.
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53)));
        for i in 0..5u8 {
            let payload = vec![i; 10];
            client
                .write_all(&datagram_frame(&target, &payload))
                .await
                .unwrap();
        }
        for i in 0..5u8 {
            let mut buf = vec![0u8; 2 + 7 + 10];
            client.read_exact(&mut buf).await.expect("reply frame");
            let (src, echoed) = decode_datagram(&buf);
            assert_eq!(src, target, "the echo reports the destination");
            assert_eq!(&echoed[..], &vec![i; 10][..], "in-order echo");
        }

        // The association is alive: another datagram round-trips.
        client
            .write_all(&datagram_frame(&target, b"alive"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 5];
        client.read_exact(&mut buf).await.expect("reply frame");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"alive");

        // Channel EOF ends the stage cleanly (the association
        // lifetime = the channel's).
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends on channel EOF")
            .expect("join");
        result.expect("clean session");
    }

    /// An empty-payload datagram is a legal datagram — never EOF (the
    /// F-2 invariant at payload level, end to end): it round-trips
    /// through the echo egress and the association survives it.
    #[tokio::test]
    async fn udp_empty_payload_datagram_is_legal_not_eof() {
        let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 2], 53)));

        // pre → empty (len = 7 inner: address only) → post, in order;
        // the empty datagram must echo as an empty datagram (a real
        // datagram), and the association must survive it.
        client
            .write_all(&datagram_frame(&target, b"pre"))
            .await
            .unwrap();
        let empty = datagram_frame(&target, b"");
        assert_eq!(
            empty.len(),
            2 + 7,
            "the empty payload's frame is addr-only inner"
        );
        client.write_all(&empty).await.unwrap();
        client
            .write_all(&datagram_frame(&target, b"post"))
            .await
            .unwrap();

        let mut buf = vec![0u8; 2 + 7 + 3];
        client.read_exact(&mut buf).await.expect("pre echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"pre");

        let mut buf = vec![0u8; 2 + 7];
        client.read_exact(&mut buf).await.expect("empty echo");
        let (src, echoed) = decode_datagram(&buf);
        assert_eq!(src, target);
        assert!(
            echoed.is_empty(),
            "the empty-payload datagram round-trips (never EOF)"
        );

        let mut buf = vec![0u8; 2 + 7 + 4];
        client.read_exact(&mut buf).await.expect("post echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"post");

        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean session");
    }

    /// A undecodable frame (garbage ATYP — any post-reply control
    /// byte parses as one) fails the association loud: EOF on the
    /// next read, no silent drop.
    #[tokio::test]
    async fn udp_undecodable_frame_ends_association() {
        let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;

        // Hand-craft a frame whose inner bytes start with an
        // unassigned ATYP.
        let inner = vec![0x7Fu8, 1, 2, 3];
        let mut garbage_frame = (inner.len() as u16).to_be_bytes().to_vec();
        garbage_frame.extend_from_slice(&inner);
        client.write_all(&garbage_frame).await.unwrap();

        // The association ends: EOF (no reply datagram follows).
        let mut buf = [0u8; 16];
        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf)).await;
        match outcome {
            Err(_) => panic!("the association must end after an undecodable frame"),
            Ok(Ok(0)) | Ok(Err(_)) => {}
            Ok(Ok(n)) => panic!("EOF expected after an undecodable frame — got {n} bytes"),
        }
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("the stage exited");
    }

    /// Channel EOF ends the association and the connection stays
    /// healthy (the POC's test 4, in-process shape): after the client
    /// half-closes one association, a fresh ASSOCIATE on a fresh
    /// duplex succeeds immediately (the same registration is
    /// unpoisoned).
    #[tokio::test]
    async fn udp_channel_eof_ends_association_and_connection_stays_healthy() {
        let egress = echo_egress_factory();
        let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await;

        // Half-close: EOF to the stage → the association ends.
        use tokio::io::AsyncWriteExt as _;
        client.shutdown().await.unwrap();
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends on channel EOF")
            .expect("join");
        result.expect("clean session (EOF is the clean exit)");
        drop(client);

        // A fresh associate succeeds immediately after (the same
        // registration, unpoisoned).
        let (handle, mut client) = raw_associate(Some(Arc::clone(&egress))).await;
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 3], 53)));
        client
            .write_all(&datagram_frame(&target, b"fresh"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 5];
        client.read_exact(&mut buf).await.expect("fresh echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"fresh");
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// The readiness contract (the POC's load-bearing API finding,
    /// re-pinned by this suite per ADR-003): an egress whose
    /// `next_reply` is an instantly-ready `Ok(None)` must not starve
    /// the stage's channel branch — datagrams still egress and echo.
    /// (The stage tolerates spurious `Ok(None)`s by re-parking.)
    #[tokio::test]
    async fn udp_instant_ok_none_egress_does_not_starve_the_stage() {
        struct HotNoneEgress {
            inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
            outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
        }

        impl crate::egress::Egress for HotNoneEgress {
            fn send(
                &mut self,
                addr: TargetAddr,
                payload: Bytes,
            ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
                Box::pin(async move {
                    let mut inner = addr
                        .to_be_bytes()
                        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
                    inner.extend_from_slice(&payload);
                    self.inbound
                        .send(Bytes::from(inner))
                        .map_err(|_| EgressError::General("egress closed".into()))
                })
            }

            fn next_reply(
                &mut self,
            ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
                Box::pin(async move {
                    // Drain one queued reply if any, then — instead of
                    // parking — return an instantly-ready `Ok(None)`
                    // (the contract violation the stage must tolerate;
                    // an actual biased-select livelock would hang this
                    // test's channel branch).
                    match self.outbound.lock().await.try_recv() {
                        Ok(inner) => Ok(Some(inner)),
                        Err(_) => Ok(None),
                    }
                })
            }
        }

        let factory: EgressFactory = Arc::new(|| {
            Box::pin(async {
                let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
                Ok(Box::new(HotNoneEgress {
                    inbound,
                    outbound: tokio::sync::Mutex::new(outbound),
                }) as Box<dyn Egress + Send>)
            })
        });

        let (handle, mut client) = raw_associate(Some(factory)).await;
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));

        // The datagrams must still egress (the channel branch
        // progresses) despite the hot `Ok(None)`s.
        client
            .write_all(&datagram_frame(&target, b"hot"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 3];
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            client.read_exact(&mut buf).await
        })
        .await
        .expect("the channel branch is not starved — the echo arrived")
        .expect("read the echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"hot");

        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// The readiness contract's zero-yield bound (the gate review's
    /// R-1, tasks/implementation/review-protocol-core.md): an
    /// egress whose `next_reply` is a pure `async { Ok(None) }` —
    /// instantly ready AND yielding no poll boundary — must still not
    /// starve the stage's channel branch. The stage's `Ok(None)`
    /// handler yields before re-parking; this test hangs (livelock)
    /// without that yield, on any scheduler.
    #[tokio::test]
    async fn udp_zero_yield_ok_none_egress_does_not_starve_the_stage() {
        struct PureHotNoneEgress {
            inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
            outbound: tokio::sync::mpsc::UnboundedReceiver<Bytes>,
        }

        impl crate::egress::Egress for PureHotNoneEgress {
            fn send(
                &mut self,
                addr: TargetAddr,
                payload: Bytes,
            ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
                Box::pin(async move {
                    let mut inner = addr
                        .to_be_bytes()
                        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
                    inner.extend_from_slice(&payload);
                    self.inbound
                        .send(Bytes::from(inner))
                        .map_err(|_| EgressError::General("egress closed".into()))
                })
            }

            fn next_reply(
                &mut self,
            ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
                Box::pin(async {
                    // Zero awaits: instantly ready, no yield point at
                    // all — the strictest readiness-contract
                    // violation (the mutex-guarded hot egress in
                    // `udp_instant_ok_none_egress_does_not_starve_
                    // the_stage` accidentally yields at its lock;
                    // this one must be survived structurally).
                    match self.outbound.try_recv() {
                        Ok(inner) => Ok(Some(inner)),
                        Err(_) => Ok(None),
                    }
                })
            }
        }

        let factory: EgressFactory = Arc::new(|| {
            Box::pin(async {
                let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
                Ok(Box::new(PureHotNoneEgress { inbound, outbound }) as Box<dyn Egress + Send>)
            })
        });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(
            server_end,
            None,
            Some(factory),
            None,
            false,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("pure.hot".into(), 1080),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00);

        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 6], 53)));
        client
            .write_all(&datagram_frame(&target, b"pure"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 4];
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            client.read_exact(&mut buf).await
        })
        .await
        .expect("the channel branch is not starved — the echo arrived")
        .expect("read the echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"pure");

        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// A missing egress callback refuses ASSOCIATE RFC-faithfully
    /// (`CommandNotSupported` 0x07 — ADR-008's missing-callback class,
    /// never a panic), and CONNECT still serves the same registration
    /// (the regression guard).
    #[tokio::test]
    async fn missing_egress_refuses_associate_command_not_supported() {
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
        let handle = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None, false));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("no.egress".into(), 1080),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x07, "missing egress → CommandNotSupported");
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::UnknownCommand(_))
        ));

        // CONNECT still serves the same registration shape (the
        // regression guard).
        let (handle, mut client) =
            raw_connect(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())))).await;
        greet(&mut client).await;
        let (code, _addr) = read_reply_after_connect(&mut client).await;
        assert_eq!(code, 0x00, "CONNECT still serves");
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// The egress-factory failure (post-command, pre-reply) maps to
    /// reply#1's code — no phantom datagram plane.
    #[tokio::test]
    async fn egress_factory_failure_maps_to_reply_code() {
        let factory: EgressFactory =
            Arc::new(|| Box::pin(async { Err(EgressError::Unreachable("no egress".into())) }));
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
        let handle = tokio::spawn(serve_socks5_over(
            server_end,
            Some(dial),
            Some(factory),
            None,
            false,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("down.egress".into(), 1080),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x03, "egress-factory refusal → NetworkUnreachable");
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("egress refused in-stream"))
        ));
    }

    /// The egress lifecycle: the factory is called once per
    /// ASSOCIATE, the stage owns the egress, and stage exit drops it
    /// (the teardown-at-birth witness — the drop flag flips when the
    /// channel EOFs, not before).
    #[tokio::test]
    async fn egress_factory_called_once_and_dropped_with_the_stage() {
        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};

        static CALLS: AtomicUsize = AtomicUsize::new(0);
        let dropped = Arc::new(AtomicBool::new(false));
        let dropped_for_factory = Arc::clone(&dropped);

        let factory: EgressFactory = Arc::new(move || {
            CALLS.fetch_add(1, Ordering::SeqCst);
            let dropped = Arc::clone(&dropped_for_factory);
            Box::pin(async move {
                let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel();
                struct Witness {
                    flag: Arc<AtomicBool>,
                    inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
                    outbound: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
                }
                impl crate::egress::Egress for Witness {
                    fn send(
                        &mut self,
                        addr: TargetAddr,
                        payload: Bytes,
                    ) -> futures::future::BoxFuture<'_, Result<(), EgressError>>
                    {
                        let mut inner = addr.to_be_bytes().unwrap().to_vec();
                        inner.extend_from_slice(&payload);
                        Box::pin(async move {
                            self.inbound
                                .send(Bytes::from(inner))
                                .map_err(|_| EgressError::General("egress closed".into()))
                        })
                    }

                    fn next_reply(
                        &mut self,
                    ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>>
                    {
                        Box::pin(async move {
                            match self.outbound.lock().await.recv().await {
                                Some(inner) => Ok(Some(inner)),
                                None => Err(EgressError::General("egress closed".into())),
                            }
                        })
                    }
                }
                impl Drop for Witness {
                    fn drop(&mut self) {
                        self.flag.store(true, Ordering::SeqCst);
                    }
                }
                Ok(Box::new(Witness {
                    flag: dropped,
                    inbound,
                    outbound: tokio::sync::Mutex::new(outbound),
                }) as Box<dyn Egress + Send>)
            })
        });

        let (handle, mut client) = raw_associate(Some(factory)).await;
        assert_eq!(CALLS.load(Ordering::SeqCst), 1, "the factory ran once");
        assert!(
            !dropped.load(Ordering::SeqCst),
            "the egress lives while the stage runs"
        );

        // One round-trip (the stage is alive).
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));
        client
            .write_all(&datagram_frame(&target, b"witness"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 7];
        client.read_exact(&mut buf).await.expect("echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"witness");

        // Channel EOF → stage exit → the egress drops.
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
        assert!(
            dropped.load(Ordering::SeqCst),
            "stage exit drops the egress (teardown-at-birth)"
        );
    }

    // -- the datagram stage's failure exits (review 002 N-8) -----------
    // The stage's teardown surface, driven directly: each failure exit
    // returns its exact `StageExit` (the serve layer discards it, so
    // the unit shape is the only place the matrix is pinnable).

    /// A stream half that fails its reads or writes on demand (the
    /// stage's `Io`-exit triggers).
    struct FailStream {
        fail_read: bool,
        fail_write: bool,
    }

    impl tokio::io::AsyncRead for FailStream {
        fn poll_read(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            _buf: &mut tokio::io::ReadBuf<'_>,
        ) -> Poll<std::io::Result<()>> {
            if self.fail_read {
                Poll::Ready(Err(std::io::Error::other("stage read failed")))
            } else {
                Poll::Pending
            }
        }
    }

    impl tokio::io::AsyncWrite for FailStream {
        fn poll_write(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &[u8],
        ) -> Poll<std::io::Result<usize>> {
            if self.fail_write {
                Poll::Ready(Err(std::io::Error::other("stage write failed")))
            } else {
                Poll::Ready(Ok(buf.len()))
            }
        }

        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }

        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
            Poll::Ready(Ok(()))
        }
    }

    /// An egress yielding one pre-set reply then parking forever (the
    /// reply branch wins the select once; the stage never advances
    /// past it).
    struct OneReplyEgress {
        reply: Option<Bytes>,
    }

    impl crate::egress::Egress for OneReplyEgress {
        fn send(
            &mut self,
            _addr: TargetAddr,
            _payload: Bytes,
        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
            Box::pin(async move { Err(EgressError::General("send refused".into())) })
        }

        fn next_reply(
            &mut self,
        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
            let reply = self.reply.take();
            Box::pin(async move {
                match reply {
                    Some(inner) => Ok(Some(inner)),
                    None => Ok(std::future::pending::<Option<Bytes>>().await),
                }
            })
        }
    }

    /// An egress whose relay already ended (`next_reply` errs) — the
    /// `EgressEnded`-via-`next_reply` trigger.
    struct RelayEndedEgress;

    impl crate::egress::Egress for RelayEndedEgress {
        fn send(
            &mut self,
            _addr: TargetAddr,
            _payload: Bytes,
        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
            Box::pin(async move { Err(EgressError::General("send refused".into())) })
        }

        fn next_reply(
            &mut self,
        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
            Box::pin(async move { Err(EgressError::General("relay ended".into())) })
        }
    }

    /// An egress that parks forever on `next_reply` (the read branch's
    /// exits are then deterministic — the parked-accept shape's egress
    /// mirror).
    struct ParkedEgress;

    impl crate::egress::Egress for ParkedEgress {
        fn send(
            &mut self,
            _addr: TargetAddr,
            _payload: Bytes,
        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
            Box::pin(async move { Err(EgressError::General("send refused".into())) })
        }

        fn next_reply(
            &mut self,
        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
            Box::pin(async move { Ok(std::future::pending::<Option<Bytes>>().await) })
        }
    }

    /// The `ReplyFrameRejected` exit: an egress reply whose inner
    /// bytes violate the u16 joint bound — the wire codec refuses it
    /// at `frame_inner`, the stage tears the association down (never a
    /// wire overflow).
    #[tokio::test]
    async fn datagram_stage_reply_frame_rejected_exit() {
        let mut egress = Box::new(OneReplyEgress {
            reply: Some(Bytes::from(vec![0u8; crate::wire::MAX_INNER_LEN + 1])),
        }) as Box<dyn Egress + Send>;
        let exit = run_datagram_stage(
            FailStream {
                fail_read: false,
                fail_write: true,
            },
            egress.as_mut(),
        )
        .await;
        assert_eq!(exit, StageExit::ReplyFrameRejected);
    }

    /// The `EgressEnded` exit via `next_reply`'s `Err` — the egress
    /// relay ended; the association ends with it (the lossy-datagram
    /// posture).
    #[tokio::test]
    async fn datagram_stage_egress_relay_ended_exit_via_next_reply() {
        let mut egress = Box::new(RelayEndedEgress) as Box<dyn Egress + Send>;
        let exit = run_datagram_stage(
            FailStream {
                fail_read: false,
                fail_write: false,
            },
            egress.as_mut(),
        )
        .await;
        assert_eq!(exit, StageExit::EgressEnded);
    }

    /// The `Io` exit via a failing write: a valid reply frame arrives
    /// but the channel's write half fails — the stage tears the
    /// association down.
    #[tokio::test]
    async fn datagram_stage_write_failure_is_the_io_exit() {
        let inner = {
            let mut v = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53)))
                .to_be_bytes()
                .unwrap()
                .to_vec();
            v.extend_from_slice(b"payload");
            Bytes::from(v)
        };
        let mut egress = Box::new(OneReplyEgress { reply: Some(inner) }) as Box<dyn Egress + Send>;
        let exit = run_datagram_stage(
            FailStream {
                fail_read: false,
                fail_write: true,
            },
            egress.as_mut(),
        )
        .await;
        assert_eq!(exit, StageExit::Io);
    }

    /// The `Io` exit via a failing channel read — the stage tears the
    /// association down (a stream error is never a silent end).
    #[tokio::test]
    async fn datagram_stage_read_failure_is_the_io_exit() {
        let mut egress = Box::new(ParkedEgress) as Box<dyn Egress + Send>;
        let exit = run_datagram_stage(
            FailStream {
                fail_read: true,
                fail_write: false,
            },
            egress.as_mut(),
        )
        .await;
        assert_eq!(exit, StageExit::Io);
    }

    /// The `EgressEnded` exit via a failing `egress.send`, at the
    /// serve layer (the drop-witness treatment
    /// `egress_factory_called_once_and_dropped_with_the_stage` gives
    /// the happy path): one datagram in, the egress refuses it, the
    /// stage exits — EOF reaches the client and the egress drops with
    /// the stage (teardown-at-birth on the failure path).
    #[tokio::test]
    async fn udp_egress_send_failure_ends_the_association_and_drops_the_egress() {
        use std::sync::atomic::{AtomicBool, Ordering};

        let dropped = Arc::new(AtomicBool::new(false));
        let dropped_for_factory = Arc::clone(&dropped);
        let factory: EgressFactory = Arc::new(move || {
            let dropped = Arc::clone(&dropped_for_factory);
            Box::pin(async move {
                struct FailingSendWitness {
                    flag: Arc<AtomicBool>,
                }
                impl crate::egress::Egress for FailingSendWitness {
                    fn send(
                        &mut self,
                        _addr: TargetAddr,
                        _payload: Bytes,
                    ) -> futures::future::BoxFuture<'_, Result<(), EgressError>>
                    {
                        Box::pin(
                            async move { Err(EgressError::General("egress send failed".into())) },
                        )
                    }

                    fn next_reply(
                        &mut self,
                    ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>>
                    {
                        Box::pin(async move { Ok(std::future::pending::<Option<Bytes>>().await) })
                    }
                }
                impl Drop for FailingSendWitness {
                    fn drop(&mut self) {
                        self.flag.store(true, Ordering::SeqCst);
                    }
                }
                Ok(Box::new(FailingSendWitness { flag: dropped }) as Box<dyn Egress + Send>)
            })
        });

        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(
            server_end,
            None,
            Some(factory),
            None,
            false,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("send.fail".into(), 1080),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "the associate reply succeeds");

        // One datagram in: the egress refuses it → the stage exits.
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));
        client
            .write_all(&datagram_frame(&target, b"doomed"))
            .await
            .unwrap();

        // The association ends: EOF (no reply datagram follows).
        let mut buf = [0u8; 16];
        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf))
                .await
                .expect("read resolves");
        match outcome {
            Ok(0) | Err(_) => {}
            Ok(n) => panic!("EOF expected after the egress send failure — got {n} bytes"),
        }
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("the stage exited");
        assert!(
            dropped.load(Ordering::SeqCst),
            "the stage exit drops the egress on the failure path too"
        );
    }

    /// CONNECT and ASSOCIATE coexist on one `serve_socks5_over` (the
    /// regression guard, extended): both commands serve their data
    /// planes over separate duplexes on the same seam set.
    #[tokio::test]
    async fn connect_and_associate_serve_on_the_same_registration() {
        let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let dial = echo_dial(Arc::clone(&targets));

        // CONNECT first.
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(serve_socks5_over(
            server_end,
            Some(Arc::clone(&dial)),
            Some(echo_egress_factory()),
            None,
            false,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("tcp.test".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00);
        client.write_all(b"tcp").await.unwrap();
        let mut buf = [0u8; 3];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"tcp");
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");

        // ASSOCIATE second — the same seams.
        let (handle, mut client) = raw_associate(Some(echo_egress_factory())).await;
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 5], 53)));
        client
            .write_all(&datagram_frame(&target, b"udp"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 3];
        client.read_exact(&mut buf).await.expect("echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"udp");
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// The CONNECT conversation's front half (the regression guard's
    /// helper: greet + request + reply).
    async fn read_reply_after_connect(client: &mut tokio::io::DuplexStream) -> (u8, [u8; 6]) {
        write_request(
            client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("echo.test".into(), 443),
        )
        .await;
        read_reply(client).await
    }

    /// The dispatch skeleton + the two-pump shape coexist: CONNECT
    /// serves, ASSOCIATE refuses, BIND is gated — one
    /// `serve_socks5_over`, all three commands (the regression
    /// guard).
    #[tokio::test]
    async fn dispatch_guard_connect_serves_while_associate_bind_refuse() {
        let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
        let (handle, mut client) = raw_connect(Arc::clone(&dial)).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("ok.test".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00);
        client.write_all(b"ok").await.unwrap();
        let mut buf = [0u8; 2];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"ok");
        drop(client);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join");
    }

    // -- BIND: the phase model (ADR-007; the POC #5 shape, in-process)
    // ---------------------------------------------------------------

    /// An accept factory whose future resolves instantly (reply#1 →
    /// reply#2 race-free for the happy-path tests).
    fn bind_accept_factory(
        requested: Arc<tokio::sync::Mutex<Option<TargetAddr>>>,
    ) -> BindAcceptFactory {
        Arc::new(move |target: TargetAddr| {
            let requested = Arc::clone(&requested);
            Box::pin(async move {
                *requested.lock().await = Some(target.clone());
                let bind_addr = SocketAddr::from(([127, 0, 0, 1], 2121));
                let accept_future = async {
                    let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024);
                    tokio::spawn(async move {
                        loop {
                            let mut buf = vec![0u8; 4096];
                            match echo_end.read(&mut buf).await {
                                Ok(0) | Err(_) => return,
                                Ok(n) => {
                                    if echo_end.write_all(&buf[..n]).await.is_err() {
                                        return;
                                    }
                                }
                            }
                        }
                    });
                    let (read_half, write_half) = tokio::io::split(target_end);
                    Ok(BindTargetHandle {
                        peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))),
                        read: Box::new(read_half) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                        write: Box::new(write_half) as Box<dyn AsyncWrite + Send + Sync + Unpin>,
                    })
                };
                Ok(BindAccept::new(bind_addr, accept_future))
            })
        })
    }

    /// An accept factory whose future parks forever (the quiet wait's
    /// other exits are then deterministic — garbage/EOF win, the
    /// accept never does): the common module's shape.
    fn parked_accept_factory() -> BindAcceptFactory {
        common::parked_accept_factory()
    }

    /// The BIND happy path through the phase model: reply#1 carries
    /// the factory's BND.ADDR, reply#2 carries the peer address, and
    /// the data plane is raw pass-through (echo both directions).
    #[tokio::test]
    async fn bind_round_trip_reply1_reply2_then_pass_through() {
        let requested = Arc::new(tokio::sync::Mutex::new(None));
        let accept = bind_accept_factory(Arc::clone(&requested));
        let (handle, mut client) = raw_connect_with(None, Some(accept), true).await;

        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("bind.test".into(), 2121),
        )
        .await;

        // Reply#1: success + the factory's advertised BND.ADDR.
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "reply#1 success");
        assert_eq!(
            SocketAddr::from((
                [addr[0], addr[1], addr[2], addr[3]],
                u16::from_be_bytes([addr[4], addr[5]])
            )),
            SocketAddr::from(([127, 0, 0, 1], 2121)),
            "reply#1 carries the factory's bind_addr"
        );
        // The factory saw the requested target (the interception
        // point — the CONNECT dial's mirror).
        assert_eq!(
            requested.lock().await.clone(),
            Some(TargetAddr::Domain("bind.test".into(), 2121))
        );

        // Reply#2: success + the accepted peer address.
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "reply#2 success");
        assert_eq!(
            SocketAddr::from((
                [addr[0], addr[1], addr[2], addr[3]],
                u16::from_be_bytes([addr[4], addr[5]])
            )),
            SocketAddr::from(([10, 0, 0, 7], 44444)),
            "reply#2 carries the accepted peer"
        );

        // Phase 2: raw pass-through, both directions (the CONNECT
        // shape — zero per-packet framing).
        client.write_all(b"ping").await.unwrap();
        let mut buf = [0u8; 4];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"ping");
        client.write_all(b"pong").await.unwrap();
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"pong");

        // Half-close teardown: client EOF ends the session cleanly.
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends on client EOF")
            .expect("join");
        result.expect("clean session");
    }

    /// The command gate (ADR-006; POC #5 test 4): BIND without
    /// `socks5:bind` gets the RFC-faithful `ConnectionNotAllowed`
    /// (0x02) — while CONNECT still serves the same registration
    /// (the gate distinguishes ingress from egress; the scopes are
    /// never implied).
    #[tokio::test]
    async fn bind_command_gate_refuses_0x02_while_connect_still_serves() {
        // The gate closed: 0x02 on BIND — no factory call (the gate
        // precedes the seam; a scope-less BIND must never mint a
        // listener).
        let requested = Arc::new(tokio::sync::Mutex::new(None));
        let accept = bind_accept_factory(Arc::clone(&requested));
        let (handle, mut client) = raw_connect_with(None, Some(accept), false).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("gate.test".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x02, "the gate refuses ConnectionNotAllowed");
        assert!(
            requested.lock().await.is_none(),
            "the gate precedes the accept seam — no phantom listener"
        );
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("bind command gate: not allowed"))
        ));

        // CONNECT still serves the same identity (egress unaffected).
        let dial = echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new())));
        let (handle, mut client) = raw_connect_with(Some(dial), None, false).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("still.ok".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "CONNECT is unaffected by the bind gate");
        drop(client);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join");
    }

    /// The quiet wait: a control byte after reply#1 is protocol
    /// garbage — the session ends with no data plane (no reply#2,
    /// no bytes echo).
    #[tokio::test]
    async fn bind_quiet_wait_garbage_byte_ends_the_session() {
        let accept = parked_accept_factory();
        let (handle, mut client) = raw_connect_with(None, Some(accept), true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("quiet.test".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "reply#1 succeeded");

        // Protocol garbage in the quiet wait (a BIND client never
        // sends another control byte).
        client.write_all(&[0x05, 0x02]).await.unwrap();

        // The session ends: EOF, no reply#2, no data plane.
        let mut buf = [0u8; 10];
        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf)).await;
        match outcome {
            Err(_) => panic!("the session must end on quiet-wait garbage"),
            Ok(Err(_)) => {}
            Ok(Ok(0)) => {}
            Ok(Ok(n)) => panic!("no reply#2 after garbage — got {n} bytes"),
        }
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("garbage surfaces as the control-garbage error");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::UnexpectedUdpControlGarbage(_))
        ));
    }

    /// Client-gone teardown: EOF mid-quiet-wait drops the accept
    /// future (the client-gone contract — whatever the provider
    /// opened dies with it). The session ends cleanly with no reply#2.
    #[tokio::test]
    async fn bind_client_gone_teardown_drops_the_accept_future() {
        // The accept future owns a listener-shaped resource: when the
        // future drops, the flag flips (the client-gone teardown made
        // observable).
        use std::sync::atomic::{AtomicBool, Ordering};
        let dropped = Arc::new(AtomicBool::new(false));
        let factory: BindAcceptFactory = {
            let dropped = Arc::clone(&dropped);
            Arc::new(move |_target: TargetAddr| {
                let dropped = Arc::clone(&dropped);
                Box::pin(async move {
                    struct Owned(Arc<AtomicBool>);
                    impl Drop for Owned {
                        fn drop(&mut self) {
                            self.0.store(true, Ordering::SeqCst);
                        }
                    }
                    let owner = Owned(Arc::clone(&dropped));
                    Ok(BindAccept::new(
                        SocketAddr::from(([127, 0, 0, 1], 2121)),
                        async move {
                            // Park forever (the wait's other branch
                            // wins); the owner's drop fires on future
                            // teardown — the client-gone contract.
                            std::future::pending::<()>().await;
                            drop(owner);
                            unreachable!("pending never resolves")
                        },
                    ))
                })
            })
        };

        let (handle, mut client) = raw_connect_with(None, Some(factory), true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("gone.test".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "reply#1 succeeded");

        // The client vanishes mid-wait: EOF → the accept future
        // drops → the owned resource tears down.
        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends on client EOF")
            .expect("join")
            .expect("clean session (EOF is not an error)");

        assert!(
            dropped.load(Ordering::SeqCst),
            "the accept future's owned resource dropped with the session"
        );
    }

    /// Factory failure (pre-reply#1) maps to reply#1's code — no
    /// phantom listener (the factory never returned one).
    #[tokio::test]
    async fn bind_factory_failure_maps_to_reply1() {
        let factory: BindAcceptFactory = Arc::new(move |_target: TargetAddr| {
            Box::pin(async move { Err(BindAcceptFactoryError::Refused("port held".into())) })
        });
        let (handle, mut client) = raw_connect_with(None, Some(factory), true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("factory.fail".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x05, "factory refusal → reply#1 ConnectionRefused");
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("bind factory refused in-stream"))
        ));
    }

    /// Accept-stage failure (post-reply#1) maps to reply#2's code —
    /// no data plane.
    #[tokio::test]
    async fn bind_accept_stage_failure_maps_to_reply2() {
        let factory: BindAcceptFactory = Arc::new(move |_target: TargetAddr| {
            Box::pin(async move {
                Ok(BindAccept::new(
                    SocketAddr::from(([127, 0, 0, 1], 2121)),
                    async move { Err(BindAcceptError::NotAllowed("policy".into())) },
                ))
            })
        });
        let (handle, mut client) = raw_connect_with(None, Some(factory), true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("accept.fail".into(), 2121),
        )
        .await;
        let (code1, _addr) = read_reply(&mut client).await;
        assert_eq!(code1, 0x00, "reply#1 succeeded — the factory resolved");
        let (code2, _addr) = read_reply(&mut client).await;
        assert_eq!(
            code2, 0x02,
            "accept-stage refusal → reply#2 ConnectionNotAllowed"
        );
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("bind accept refused in-stream"))
        ));
    }

    /// The no-factory default (ADR-007 variant 3, wrapper-aware): the
    /// sentinel reply#1 + the sentinel reply#2 (`0.0.0.0:0`), the
    /// full phase model preserved end to end (quiet wait included —
    /// garbage still ends the session), then raw pass-through (the
    /// in-band halves are empty/sink — bytes flow one way to the
    /// sink, EOF comes back).
    #[tokio::test]
    async fn bind_no_factory_serves_the_wrapper_aware_variant_3() {
        let (handle, mut client) = raw_connect_with(None, None, true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("variant3.test".into(), 2121),
        )
        .await;
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "variant 3 serves reply#1");
        assert_eq!(
            (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
            (0, 0, 0, 0, 0, 0),
            "reply#1 carries the wrapper-aware sentinel"
        );
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "variant 3 serves reply#2");
        assert_eq!(
            (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
            (0, 0, 0, 0, 0, 0),
            "reply#2 carries the wrapper-aware sentinel peer"
        );
        // The data plane exists (the channel is the connection): the
        // in-band halves are empty/sink, so the client's write lands
        // and EOF follows.
        client.write_all(b"hi").await.unwrap();
        let mut eof = [0u8; 1];
        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut eof))
                .await
                .expect("read resolves");
        assert_eq!(outcome.unwrap_or(0), 0, "the empty half EOFs");
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean session");
    }

    /// The establishment bound covers only the open-op round trip
    /// (ADR-007): the quiet wait outlives any establishment-length
    /// timeout — a client that BINDs and stays silent (no dial) keeps
    /// the session alive (the wait is teardown-governed; the
    /// over-channels shape is the harness's pin).
    #[tokio::test]
    async fn bind_quiet_wait_is_not_under_the_establishment_bound() {
        let accept = parked_accept_factory();
        let (handle, mut client) = raw_connect_with(None, Some(accept), true).await;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("ftp.wait".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "reply#1 succeeded");

        // Stay silent — the wait is teardown-governed, not
        // establishment-governed. (The substrate-free shape has no
        // alkcall bound; the harness integration test pins the
        // over-channels shape. Here the parked accept still answers
        // after a sleep — the wait is alive, not under any in-stream
        // timeout.)
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        client.write_all(b"x").await.unwrap();
        let mut buf = [0u8; 1];
        let outcome =
            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf)).await;
        match outcome {
            Err(_) => panic!("the session must end on quiet-wait garbage"),
            Ok(Ok(0)) | Ok(Err(_)) => {}
            Ok(Ok(n)) => panic!("no reply#2 after garbage — got {n} bytes"),
        }
        drop(client);
        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join");
    }

    /// The quiet wait's `Err(_)` arm (review 002 N-8): a channel read
    /// that *errors* (I/O error, not EOF) during the quiet wait ends
    /// the wait with a clean `Ok(())` — the accept future drops with
    /// the session (the client-gone contract's error-shaped twin).
    /// Driven directly on `bind_quiet_wait_and_serve` (a read error is
    /// not producible over a duplex).
    #[tokio::test]
    async fn bind_quiet_wait_read_error_ends_the_wait_cleanly() {
        use std::cell::Cell;
        use std::sync::atomic::{AtomicBool, Ordering};

        let dropped = Arc::new(AtomicBool::new(false));
        struct FailReadStream {
            polled: Cell<bool>,
        }
        impl tokio::io::AsyncRead for FailReadStream {
            fn poll_read(
                self: Pin<&mut Self>,
                cx: &mut Context<'_>,
                _buf: &mut tokio::io::ReadBuf<'_>,
            ) -> Poll<std::io::Result<()>> {
                // Park once (both branches get polled — the accept
                // future starts, its owned state exists), then error
                // on the wake: the read branch's error resolves after
                // the accept branch has started.
                if self.polled.get() {
                    Poll::Ready(Err(std::io::Error::other("quiet-wait read failed")))
                } else {
                    self.polled.set(true);
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            }
        }
        impl tokio::io::AsyncWrite for FailReadStream {
            fn poll_write(
                self: Pin<&mut Self>,
                _cx: &mut Context<'_>,
                buf: &[u8],
            ) -> Poll<std::io::Result<usize>> {
                Poll::Ready(Ok(buf.len()))
            }
            fn poll_flush(
                self: Pin<&mut Self>,
                _cx: &mut Context<'_>,
            ) -> Poll<std::io::Result<()>> {
                Poll::Ready(Ok(()))
            }
            fn poll_shutdown(
                self: Pin<&mut Self>,
                _cx: &mut Context<'_>,
            ) -> Poll<std::io::Result<()>> {
                Poll::Ready(Ok(()))
            }
        }

        struct OwnedAccept(Arc<AtomicBool>);
        impl Drop for OwnedAccept {
            fn drop(&mut self) {
                self.0.store(true, Ordering::SeqCst);
            }
        }
        let accept = {
            let dropped = Arc::clone(&dropped);
            BindAccept::new(SocketAddr::from(([127, 0, 0, 1], 2121)), async move {
                let owner = OwnedAccept(dropped);
                std::future::pending::<()>().await;
                drop(owner);
                unreachable!("pending never resolves")
            })
        };

        let outcome = bind_quiet_wait_and_serve(
            FailReadStream {
                polled: Cell::new(false),
            },
            accept,
        )
        .await;
        outcome.expect("an error-shaped quiet wait is not an error exit");
        assert!(
            dropped.load(Ordering::SeqCst),
            "the accept future dropped with the wait (the client-gone contract, error arm)"
        );
    }

    /// The BIND reply#2 Domain-peer fallback (review 002 N-8): a
    /// foreign accept handle carrying a `TargetAddr::Domain(_, port)`
    /// peer — the producer writes reply#2 with the unspecified
    /// address on the peer's port (`0.0.0.0:port`) and serves the
    /// data plane.
    #[tokio::test]
    async fn bind_domain_peer_falls_back_to_the_unspecified_address() {
        let factory: BindAcceptFactory = Arc::new(move |_target: TargetAddr| {
            Box::pin(async move {
                Ok(BindAccept::new(
                    SocketAddr::from(([127, 0, 0, 1], 2121)),
                    async {
                        let (mut echo_end, target_end) = tokio::io::duplex(64);
                        tokio::spawn(async move {
                            let mut buf = [0u8; 64];
                            loop {
                                match echo_end.read(&mut buf).await {
                                    Ok(0) | Err(_) => return,
                                    Ok(n) => {
                                        if echo_end.write_all(&buf[..n]).await.is_err() {
                                            return;
                                        }
                                    }
                                }
                            }
                        });
                        let (read_half, write_half) = tokio::io::split(target_end);
                        Ok(BindTargetHandle {
                            peer: TargetAddr::Domain("peer.example".into(), 44444),
                            read: Box::new(read_half) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                            write: Box::new(write_half)
                                as Box<dyn AsyncWrite + Send + Sync + Unpin>,
                        })
                    },
                ))
            })
        });
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let handle = tokio::spawn(serve_socks5_over(
            server_end,
            None,
            None,
            Some(factory),
            true,
        ));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("domain.peer".into(), 2121),
        )
        .await;

        // Reply#1 succeeds; reply#2 carries the Domain peer's fallback
        // (the unspecified address, the peer's port).
        let (code1, _addr) = read_reply(&mut client).await;
        assert_eq!(code1, 0x00, "reply#1 succeeded");
        let (code2, addr) = read_reply(&mut client).await;
        assert_eq!(code2, 0x00, "reply#2 succeeded");
        assert_eq!(
            SocketAddr::from((
                [addr[0], addr[1], addr[2], addr[3]],
                u16::from_be_bytes([addr[4], addr[5]])
            )),
            SocketAddr::from(([0, 0, 0, 0], 44444)),
            "the Domain peer fell back to 0.0.0.0:port"
        );

        // The data plane is live (the accept resolved with in-process
        // halves — the parked probe half stays alive by forget).
        client.write_all(b"ping").await.unwrap();
        let mut buf = [0u8; 4];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            client.read_exact(&mut buf),
        )
        .await
        .expect("the data plane echoes")
        .expect("read the echo");
        assert_eq!(&buf, b"ping");

        drop(client);
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("session ends")
            .expect("join")
            .expect("clean");
    }

    /// The pump handler's shape: `accept_bi` → the same conversation,
    /// with the handler's `JoinHandle` tracking the data plane (R-02
    /// — the spawn shape is inline-await, not spawn-and-forget).
    #[tokio::test]
    async fn pump_handler_tracks_the_data_plane_to_completion() {
        let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let dial = echo_dial(Arc::clone(&targets));

        let handler = make_socks5_pump_handler(Some(Arc::clone(&dial)), None, None);

        // A fake alkcall `Connection` is not constructible outside
        // alkcall; drive the substrate-free path instead — the
        // handler's task body is exactly `accept_bi` +
        // `serve_socks5_over` (the shape this test pins is the
        // inline-await). The full handler-over-channels shape is the
        // harness's integration tests.
        let _ = &handler;
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(serve_socks5_over(server_end, Some(dial), None, None, false));
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("inline.test".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00);
        client.write_all(b"tick").await.unwrap();
        let mut buf = [0u8; 4];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"tick");
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("the session ends on client EOF — the data plane is tracked")
            .expect("join");
        result.expect("clean session");
        assert_eq!(
            targets.lock().await.last(),
            Some(&TargetAddr::Domain("inline.test".into(), 443))
        );
    }

    /// The establisher records the opener identity (CF-006) and
    /// returns the no-plan establishment (the session gate — the
    /// target is unknown at open time, POC #1's wire note).
    #[tokio::test]
    async fn establisher_records_identity_and_defaults_the_establishment() {
        let witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None));
        let establisher = socks5_session_establisher(Some(Arc::clone(&witness)));
        let auth = AuthContext {
            identity: Some(alkcall::core::auth::Identity {
                id: "opener".to_string(),
                scopes: vec![SOCKS5_OPEN_SCOPE.to_string()],
                resources: Default::default(),
            }),
            ..AuthContext::anonymous(b"alk/socks5")
        };
        let establishment = establisher(serde_json::json!({}), auth)
            .await
            .expect("the session gate passes");
        assert!(establishment.plan.is_none(), "no plan payload");
        assert_eq!(
            witness.lock().await.as_deref(),
            Some("opener"),
            "the per-call opener identity is recorded"
        );

        // The None-witness arm (the registration's optional witness
        // seam): the session gate passes without recording anything.
        let establisher = socks5_session_establisher(None);
        let auth = AuthContext::anonymous(b"alk/socks5");
        let establishment = establisher(serde_json::json!({}), auth)
            .await
            .expect("the session gate passes without a witness");
        assert!(establishment.plan.is_none(), "no plan payload");
    }

    /// The command gate's resolved boolean rides the open-op
    /// `AuthContext`'s identity scopes (CF-005/CF-006, ADR-006): the
    /// gate is `socks5:bind`-scoped — never implied by `socks5:open`.
    #[test]
    fn pump_handler_resolves_the_bind_gate_from_the_opener_identity() {
        fn gate(scopes: &[&str]) -> bool {
            let auth = AuthContext {
                identity: Some(alkcall::core::auth::Identity {
                    id: "caller".to_string(),
                    scopes: scopes.iter().map(|s| s.to_string()).collect(),
                    resources: Default::default(),
                }),
                ..AuthContext::anonymous(b"alk/socks5")
            };
            auth.identity
                .as_ref()
                .map(|i| i.scopes.iter().any(|s| s == SOCKS5_BIND_SCOPE))
                .unwrap_or(false)
        }

        assert!(!gate(&[SOCKS5_OPEN_SCOPE]), "open does not imply bind");
        assert!(gate(&[SOCKS5_BIND_SCOPE]));
        assert!(gate(&[SOCKS5_OPEN_SCOPE, SOCKS5_BIND_SCOPE]));
        assert!(!gate(&[]), "scope-less fails closed");
    }

    /// The builder assembly shape (producer.md §Wiring): seams ride
    /// the builder; `serve_socks5_over` over the built producer
    /// behaves like the free function.
    #[tokio::test]
    async fn builder_shape_round_trips() {
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .build();
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("builder.test".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00);
        client.write_all(b"ping").await.unwrap();
        let mut buf = [0u8; 4];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"ping");
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean");
    }

    /// The gate policy (review 002 C-3): the builder's `bind_policy`
    /// is the substrate-free entry's BIND gate — "factory wired but
    /// gate closed" is expressible (`bind_policy(false)` with the
    /// accept seam still wired refuses BIND 0x02, no factory call —
    /// the gate precedes the seam) while CONNECT and ASSOCIATE still
    /// serve the same producer. Seam presence is never the sole gate.
    #[tokio::test]
    async fn builder_bind_policy_false_refuses_bind_while_connect_associate_serve() {
        let requested = Arc::new(tokio::sync::Mutex::new(None));
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .egress(echo_egress_factory())
            .accept(bind_accept_factory(Arc::clone(&requested)))
            .bind_policy(false)
            .build();

        // BIND with the factory wired and the gate closed: 0x02, and
        // the accept seam is never called (no phantom listener — the
        // gate precedes the seam, the ADR-006 shape).
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("policy.closed".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(
            code, 0x02,
            "the explicit policy refuses ConnectionNotAllowed"
        );
        assert!(
            requested.lock().await.is_none(),
            "the closed gate precedes the accept seam — no phantom listener"
        );
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("bind command gate: not allowed"))
        ));

        // CONNECT still serves the same producer (egress unaffected).
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .egress(echo_egress_factory())
            .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None))))
            .bind_policy(false)
            .build();
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPConnect,
            &TargetAddr::Domain("policy.connect".into(), 443),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "CONNECT is unaffected by the bind policy");
        client.write_all(b"ping").await.unwrap();
        let mut buf = [0u8; 4];
        client.read_exact(&mut buf).await.unwrap();
        assert_eq!(&buf, b"ping");
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean");

        // ASSOCIATE still serves the same producer.
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .egress(echo_egress_factory())
            .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None))))
            .bind_policy(false)
            .build();
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::UDPAssociate,
            &TargetAddr::Domain("policy.udp".into(), 1080),
        )
        .await;
        let (code, addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "ASSOCIATE is unaffected by the bind policy");
        assert_eq!(
            (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
            (127, 0, 0, 1, 0, 0),
            "the reply carries the sentinel BND.ADDR"
        );
        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 8], 53)));
        client
            .write_all(&datagram_frame(&target, b"policy"))
            .await
            .unwrap();
        let mut buf = vec![0u8; 2 + 7 + 6];
        client.read_exact(&mut buf).await.expect("echo");
        let (_, echoed) = decode_datagram(&buf);
        assert_eq!(&echoed[..], b"policy");
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean");
    }

    /// The backward-compat default (review 002 C-3): an unwired
    /// `bind_policy` resolves the gate from the accept seam's
    /// presence — a factory-wired producer serves BIND on the
    /// substrate-free entry (the landed tests' reading preserved),
    /// while a producer without the factory still refuses. The
    /// explicit policy always wins when wired.
    #[tokio::test]
    async fn builder_default_bind_policy_follows_accept_presence() {
        // Accept wired, policy unwired → BIND serves (the default).
        let requested = Arc::new(tokio::sync::Mutex::new(None));
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .accept(bind_accept_factory(Arc::clone(&requested)))
            .build();
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("default.open".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x00, "the presence default serves BIND");
        assert!(
            requested.lock().await.is_some(),
            "the accept seam ran (the factory is wired and the gate open)"
        );
        drop(client);
        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join");
        result.expect("clean");

        // Accept unwired, policy unwired → BIND refuses (the gate's
        // no-grant answer — no seam to run even if the gate were
        // open).
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .build();
        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
        let mut client = client_end;
        greet(&mut client).await;
        write_request(
            &mut client,
            &Socks5Command::TCPBind,
            &TargetAddr::Domain("default.closed".into(), 2121),
        )
        .await;
        let (code, _addr) = read_reply(&mut client).await;
        assert_eq!(code, 0x02, "no factory + no policy = the closed default");
        drop(client);
        let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
            .await
            .expect("session ends")
            .expect("join")
            .expect_err("the refusal surfaces in-stream");
        assert!(matches!(
            err,
            SocksError::ServerError(SocksServerError::Refused("bind command gate: not allowed"))
        ));
    }

    /// The sentinel reply address (the in-band BND.ADDR posture,
    /// ADR-003): `127.0.0.1:0` — the RFC-compatibility surface.
    #[test]
    fn sentinel_reply_addr_shape() {
        assert_eq!(sentinel_reply_addr(), SocketAddr::from(([127, 0, 0, 1], 0)));
    }

    /// The pump handler's seam-closure shape: `Clone`-carried
    /// `Option` seams survive the closure move (the registration's
    /// shape — the handler is reusable across sessions).
    #[test]
    fn producer_builder_seams_survive_clone() {
        let producer = Socks5Producer::builder()
            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
            .build();
        let cloned = producer.clone();
        let (client_end, server_end) = tokio::io::duplex(64);
        drop(client_end);
        drop(cloned.serve_socks5_over(server_end));
    }
}