alksocks 0.3.0

//! The consumer half: the channels-native session types and teardown
//! ownership (consumer.md §The Channels-Native Sessions).
//!
//! [`Socks5ChannelSession`] is the CONNECT session: it opens
//! `channels/socks5/sub` via `ChannelClient::open_channel` (trivial
//! params — ADR-001), drives the forked client typestate's
//! `use_stream` path over the channel's `BiStream` (POC #1's fit —
//! `Socks5Stream` impls `AsyncRead + AsyncWrite` itself, so a session
//! over a `BiStream` is the intended use, not a hack), and after the
//! reply presents the raw byte pipe (0 B framing — the session
//! reads/writes the target's bytes directly). [`BindSession`] is the
//! BIND session: the reply#2 reader (consumer.md §The BIND client,
//! ADR-007) — reply#1 rides the forked client `request()`, reply#2 is
//! hand-read with the forked `read_address` codec (no second codec),
//! then the pass-through data plane. [`AssociateSession`] is the UDP
//! ASSOCIATE session (POC #2's shape — see its docs).
//!
//! The BIND session's reply#2 phase is take-guarded (review 002
//! C-1): the first `await_inbound` takes the stream — the plane owns
//! it, the session stays the teardown owner — and a second call
//! returns the family's spent shape (`Ok(None)`); the reply#2 reader
//! is that one method alone (`reply1()` is a view, D-1).
//!
//! Target construction rides [`ToTargetAddr`]: IP-string forms parse
//! to IP targets; the domain form passes through unresolved (ADR-012
//! — resolution is the egress's job, never the protocol's).
//!
//! [`AssociateSession`] is the UDP ASSOCIATE session (POC #2's shape,
//! consumer.md §The associate session, ADR-003): it opens the
//! channel, negotiates, sends ASSOCIATE, reads the reply — **receives
//! the sentinel `BND.ADDR` and never uses it for addressing** (the
//! handle is the channel). The datagram plane is the framed stage:
//! `send_datagram` frames once (`[len: u16 BE][addr + data]` — the
//! framing-once mirror, ADR-003) and `recv_datagram` decodes
//! incrementally with **reader state that persists across calls**
//! (mid-frame position + batched-datagram pending queue — a datagram
//! split across channel chunks survives a caller-driven receive
//! loop; POC #2's finding: the first cut dropped the reader state
//! per call, the 2-byte-prefix chunk test exposed it). `shutdown`
//! half-closes the stream — the mux's implicit-EOF sentinel reaches
//! the producer's stage as EOF and the association ends (association
//! lifetime = channel lifetime).
//!
//! Teardown ownership (alktunnels ADR-005 posture, inherited via
//! ADR-004): the session owns the adopted channel entry. On Drop it
//! reaps it (`ChannelManager::teardown_channel` — sync, best-effort;
//! `Drop` cannot await) and the data plane EOFs structurally (the
//! mux's implicit-EOF sentinel on the dropped write half). The
//! teardown-at-birth discipline covers the in-stream refusal path:
//! a refused ASSOCIATE tears the adopted channel down before the
//! error returns — a refused session never leaves a phantom channel.
//! The session does **not** implement `Clone` (compile-asserted on
//! the struct): aliasing a session would alias its teardown and its
//! reader state.
//!
//! Error surface (consumer.md §Error surface): open failures are the
//! typed [`SocksOpenError`] reasons (ADR-049 §4 — a refused session
//! never yields a phantom channel); a refused ASSOCIATE is an RFC
//! reply code surfaced through [`SocksSessionError::Rfc`].

use alkcall::channels::client::{ChannelClient, ChannelOpenError};
use alkcall::channels::manager::ChannelManager;
use alkcall::core::types::BiStream;
use alkcall::protocol::connection::CallConnection;
use bytes::Bytes;
use std::net::{IpAddr, Ipv6Addr, SocketAddr};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};

use crate::error::{SocksOpenError, SocksSessionError};
use crate::params::{Socks5Params, OP_SOCKS5_OPEN, SOCKS5_ALPN};
use crate::rfc1928::client::{Config, Socks5Stream};
use crate::rfc1928::consts;
use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr};
use crate::rfc1928::{ReplyError, Socks5Command, SocksError, UdpHeaderError};
use crate::wire::{frame, DatagramReader, WireError};

/// The three session constructors' shared prologue (review 002 N-6):
/// open the channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)`
/// — the trivial params, ADR-001), join the halves into the channel's
/// `BiStream`, run the command negotiation over the forked client
/// typestate, and on any negotiation failure reap the adopted channel
/// before the error returns (teardown-at-birth — the phantom-channel
/// property, in one place instead of three). `Ok` yields the channel
/// id, the negotiated stream, and the cloned manager for the session
/// to own.
async fn open_and_negotiate<T, F, Fut>(
    client: &ChannelClient,
    negotiate: F,
) -> Result<(u32, T, ChannelManager), SocksSessionError>
where
    F: FnOnce(Socks5Stream<BiStream>) -> Fut,
    Fut: std::future::Future<Output = Result<T, SocksSessionError>>,
{
    let (channel_id, send, recv) = client
        .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
        .await
        .map_err(SocksOpenError::from)?;

    let bidi = BiStream::from_joined(recv, send);
    let negotiated = negotiate(Socks5Stream::use_stream(bidi, None, Config::default()).await?);

    match negotiated.await {
        Ok(inner) => Ok((channel_id, inner, client.manager().clone())),
        Err(err) => {
            let _ = client.manager().teardown_channel(channel_id);
            Err(err)
        }
    }
}

/// The cred-bearing front door's re-origination prologue (the
/// token-carrying twin of [`open_and_negotiate`]): open the channel
/// with the payload `auth_token` riding the `call.requested` payload
/// (CF-005 precedence (1) — the producer-side dispatch resolves the
/// token through its `ServingConfig.identity_provider` and runs the
/// ACL gate against the mapped identity), join the halves, run the
/// command negotiation, and reap on failure — the same phantom-channel
/// property. `Ok` yields the channel id, the negotiated stream, and
/// the cloned manager.
///
/// The token path needs the raw `call.requested` payload (`auth_token`
/// is a payload-level field, ADR-017 §7), so the open rides
/// `CallConnection::call_with_payload` + `adopt_channel` — the same
/// two-step `open_channel` performs, with the token added. The
/// client's call connection is taken once and cached in the returned
/// client (the door's `TokenClient` owns it for the door's lifetime).
async fn open_with_token_and_negotiate<T, F, Fut>(
    client: &crate::consumer::TokenClient,
    token: Vec<u8>,
    negotiate: F,
) -> Result<(u32, T, ChannelManager), SocksSessionError>
where
    F: FnOnce(Socks5Stream<BiStream>) -> Fut,
    Fut: std::future::Future<Output = Result<T, SocksSessionError>>,
{
    let channel_id = client.open_with_token(token).await?;

    let (send, recv) = client
        .manager()
        .adopt_channel(channel_id, SOCKS5_ALPN, None)
        .await
        .map_err(|e| SocksOpenError::from(ChannelOpenError::AdoptFailed(e)))?;
    let bidi = BiStream::from_joined(recv, send);

    let negotiated = negotiate(Socks5Stream::use_stream(bidi, None, Config::default()).await?);

    match negotiated.await {
        Ok(inner) => Ok((channel_id, inner, client.manager().clone())),
        Err(err) => {
            let _ = client.manager().teardown_channel(channel_id);
            Err(err)
        }
    }
}

/// The spent-session marker (the gate review's R-2,
/// tasks/implementation/review-protocol-core.md): [`Socks5ChannelSession
/// ::into_inner`] / [`AssociateSession::into_inner`] return it when
/// the pipe/stream was already taken — reachable only by calling
/// twice on a session whose inner is gone. Carries nothing: a spent
/// session's teardown is already done (the reap ran with the take).
#[derive(Debug, thiserror::Error)]
#[error("the session's pipe/stream was already taken (spent)")]
pub struct SpentSessionError;

/// A completed SOCKS5 CONNECT session on the consumer side: the
/// adopted channel + the post-reply byte pipe to the target.
///
/// The session is a handle to one channel (one session = one channel
/// = one CONNECT, convention 10). It does **not** implement `Clone`:
///
/// ```compile_fail
/// use alksocks::consumer::Socks5ChannelSession;
/// fn assert_clone<T: Clone>() {}
/// assert_clone::<Socks5ChannelSession>();
/// ```
///
/// The inner stream is `Option` so `Drop`/`close` can empty the
/// type (the alktunnels `Drop`-impl posture); a spent session drops
/// as a teardown-only no-op.
pub struct Socks5ChannelSession {
    channel_id: u32,
    manager: Option<ChannelManager>,
    inner: Option<Socks5Stream<BiStream>>,
    /// The CONNECT reply's BND.ADDR — captured at `connect` (the
    /// forked client `request()` returns it once; the front door's
    /// re-origination re-encodes the vanilla reply from it — the RFC
    /// codec at the door boundary only, ADR-003).
    reply_addr: Option<TargetAddr>,
}

impl std::fmt::Debug for Socks5ChannelSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Socks5ChannelSession")
            .field("channel_id", &self.channel_id)
            .field("connected", &self.inner.is_some())
            .finish()
    }
}

impl Socks5ChannelSession {
    /// CONNECT to `target` through a produced SOCKS5 resource: opens
    /// the channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)`
    /// — the trivial params, ADR-001), drives the no-auth handshake
    /// and the CONNECT request in-stream over the channel's
    /// `BiStream`, and returns the session owning both.
    ///
    /// Error surface: an open failure is [`SocksSessionError::Open`]
    /// (the typed ADR-049 reasons; no channel was ever returned); a
    /// refused target surfaces the RFC reply code through
    /// [`SocksSessionError::Rfc`] — and the adopted channel is torn
    /// down before the error returns (no phantom channel).
    pub async fn connect(
        client: &ChannelClient,
        target: impl ToTargetAddr,
    ) -> Result<Self, SocksSessionError> {
        let target: TargetAddr = target.to_target_addr()?;
        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
            let reply_addr = inner.request(Socks5Command::TCPConnect, target).await?;
            Ok((reply_addr, inner))
        })
        .await?;
        let (reply_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            reply_addr: Some(reply_addr),
        })
    }

    /// The cred-bearing front door's CONNECT re-origination (the
    /// token-carrying twin of [`Self::connect`]): the open rides the
    /// payload `auth_token` (CF-005 precedence (1) — the producer's
    /// identity provider resolves the door's mapped identity and its
    /// scopes gate the open), then the same in-stream negotiation over
    /// the channel's `BiStream`. Error surface identical.
    pub async fn connect_with_token(
        client: &TokenClient,
        target: impl ToTargetAddr,
        token: Vec<u8>,
    ) -> Result<Self, SocksSessionError> {
        let target: TargetAddr = target.to_target_addr()?;
        let (channel_id, inner, manager) =
            open_with_token_and_negotiate(client, token, |mut inner| async move {
                let reply_addr = inner.request(Socks5Command::TCPConnect, target).await?;
                Ok((reply_addr, inner))
            })
            .await?;
        let (reply_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            reply_addr: Some(reply_addr),
        })
    }

    /// The adopted channel's id (observability + correlation; one
    /// session = one channel).
    pub fn channel_id(&self) -> u32 {
        self.channel_id
    }

    /// The CONNECT reply's BND.ADDR — the in-band reply's bound
    /// address, captured at `connect` (the forked client `request()`
    /// returns it once). The front door's re-origination re-encodes
    /// the vanilla reply from it (the RFC codec at the door boundary
    /// only, ADR-003); `None` after the pipe was taken (the
    /// spent-session posture).
    pub fn reply_addr(&self) -> Option<TargetAddr> {
        self.reply_addr.clone()
    }

    /// The post-CONNECT pipe as split read/write halves (0 B framing
    /// — raw bytes to/from the target). The halves borrow the
    /// session: teardown stays owned (`close`/`Drop`). `None` after
    /// [`Self::into_inner`] took the pipe.
    pub fn split(&mut self) -> Option<SplitPipe<'_>> {
        let inner = self.inner.as_mut()?;
        Some(tokio::io::split(inner))
    }

    /// Take the underlying [`Socks5Stream<BiStream>`] (advanced use —
    /// the forked client type's socket accessors, owned-half pumps).
    /// The session rides along still owning the channel's teardown:
    /// drop the pair (or `close` the session) when done — dropping
    /// `inner` EOFs the data plane structurally.
    ///
    /// [`SpentSessionError`] on a spent session (the pipe already
    /// taken): a spent session's manager entry is always reaped
    /// already (`close`/`shutdown`/`Drop` all take it), so dropping
    /// it loses nothing — no panic path (AGENTS.md convention 2).
    pub fn into_inner(mut self) -> Result<TakenInner, SpentSessionError> {
        match self.inner.take() {
            Some(inner) => Ok(TakenInner {
                session: self,
                inner,
            }),
            None => Err(SpentSessionError),
        }
    }

    /// Tear the session down gracefully: half-close the pipe (the
    /// EOF sentinel reaches the producer's pump), then reap the
    /// adopted channel entry. Returns whether the entry existed.
    pub async fn close(mut self) -> bool {
        if let Some(inner) = self.inner.as_mut() {
            let _ = inner.shutdown().await;
        }
        self.inner.take();
        self.reap()
    }

    /// The adopted channel's teardown (`teardown_channel` returns a
    /// producer-side pump-task handle the caller may abort after the
    /// drain — the channels-call path's shape). The handle is
    /// deliberately dropped here: the producer-side pump finishes via
    /// EOF propagation (the graceful shape — aborting would cut
    /// mid-flight data); `clear_all` aborts instead (the substrate's
    /// abrupt path).
    fn reap(&mut self) -> bool {
        self.manager
            .take()
            .map(|m| m.teardown_channel(self.channel_id).is_ok())
            .unwrap_or(false)
    }
}

impl Drop for Socks5ChannelSession {
    /// Sync-reap the adopted channel entry (`teardown_channel` is
    /// sync — best-effort; `Drop` cannot await) and drop the pipe
    /// (the mux's implicit-EOF sentinel EOFs the producer's pump).
    /// Dropping without `close` never leaks the channel entry.
    fn drop(&mut self) {
        self.inner.take();
        let _ = self.reap();
    }
}

/// The UDP ASSOCIATE consumer session (POC #2's shape, consumer.md
/// §The associate session, ADR-003): the channel is the relay handle
/// — the reply's sentinel `BND.ADDR` is received and never used for
/// addressing, the datagram plane is the framed stage on the channel
/// (`[len: u16 BE][addr + data]`, both directions), and the
/// association's lifetime is the channel's.
///
/// The session is a handle to one channel (one session = one channel
/// = one ASSOCIATE, convention 10) and owns its reader state: the
/// incremental [`DatagramReader`] (mid-frame position) and the
/// pending-queue of batched datagrams both **persist across
/// `recv_datagram` calls** — a datagram split across channel chunks
/// survives a caller-driven receive loop (POC #2's finding, the
/// first cut dropped the reader state per call). It does **not**
/// implement `Clone`:
///
/// ```compile_fail
/// use alksocks::consumer::AssociateSession;
/// fn assert_clone<T: Clone>() {}
/// assert_clone::<AssociateSession>();
/// ```
///
/// The inner stream is `Option` so `Drop`/`shutdown`/`close` can
/// empty the type (the alktunnels `Drop`-impl posture); a spent
/// session drops as a teardown-only no-op.
pub struct AssociateSession {
    channel_id: u32,
    manager: Option<ChannelManager>,
    inner: Option<Socks5Stream<BiStream>>,
    /// The incremental length-codec decoder — its mid-frame state
    /// persists across `recv_datagram` calls (a datagram may be split
    /// across channel chunks; the same `DatagramReader` the producer
    /// stage runs).
    reader: DatagramReader,
    /// Datagrams decoded beyond the one returned (a chunk may batch
    /// several datagrams — drained by later `recv_datagram` calls).
    /// The batched datagrams are stored decoded (`(addr, payload)` —
    /// the same shape `recv_datagram` returns).
    pending: std::collections::VecDeque<(TargetAddr, Bytes)>,
    /// The reply's BND.ADDR — received for RFC fidelity, unused for
    /// addressing (wrapper-aware clients sendto the channel; ADR-003).
    relay_addr: Option<TargetAddr>,
}

impl std::fmt::Debug for AssociateSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AssociateSession")
            .field("channel_id", &self.channel_id)
            .field("associated", &self.inner.is_some())
            .field("relay_addr", &self.relay_addr)
            .field("pending_datagrams", &self.pending.len())
            .finish()
    }
}

impl AssociateSession {
    /// ASSOCIATE through a produced SOCKS5 resource: opens the
    /// channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)` — the
    /// trivial params, ADR-001), drives the no-auth handshake and the
    /// UDP ASSOCIATE request in-stream over the channel's `BiStream`,
    /// and reads the reply. The DST.ADDR of the ASSOCIATE command
    /// mirrors the vanilla shape — the unspecified address (`[::]:0`),
    /// as the forked `Socks5Datagram` sends. The reply's BND.ADDR is
    /// the producer's sentinel (`127.0.0.1:0` on the in-band path) —
    /// captured as [`Self::relay_addr`], never used for addressing
    /// (ADR-003: the handle is the channel).
    ///
    /// Error surface: an open failure is [`SocksSessionError::Open`]
    /// (the typed ADR-049 reasons; no channel was ever returned); a
    /// refused ASSOCIATE (the missing-egress `CommandNotSupported`
    /// 0x07 or a factory failure's mapped code) surfaces the RFC
    /// reply code through [`SocksSessionError::Rfc`] — and the
    /// adopted channel is torn down before the error returns (no
    /// phantom channel).
    pub async fn associate(client: &ChannelClient) -> Result<Self, SocksSessionError> {
        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
            let relay_addr = inner
                .request(
                    Socks5Command::UDPAssociate,
                    TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)),
                )
                .await?;
            Ok((relay_addr, inner))
        })
        .await?;
        let (relay_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            reader: DatagramReader::new(),
            pending: std::collections::VecDeque::new(),
            relay_addr: Some(relay_addr),
        })
    }

    /// The cred-bearing front door's ASSOCIATE re-origination (the
    /// token-carrying twin of [`Self::associate`]): the open rides
    /// the payload `auth_token` (CF-005 precedence (1)). Not called
    /// by the landed TCP door — its ASSOCIATE arm refuses
    /// RFC-faithfully before any session is opened (the datagram
    /// plane is the UDP door's task, ADR-003); the constructor is
    /// the session family's completeness (every command carries the
    /// token twin) and the wrapper-aware deployment's surface. The
    /// session is otherwise identical.
    pub async fn associate_with_token(
        client: &TokenClient,
        token: Vec<u8>,
    ) -> Result<Self, SocksSessionError> {
        let (channel_id, inner, manager) =
            open_with_token_and_negotiate(client, token, |mut inner| async move {
                let relay_addr = inner
                    .request(
                        Socks5Command::UDPAssociate,
                        TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)),
                    )
                    .await?;
                Ok((relay_addr, inner))
            })
            .await?;
        let (relay_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            reader: DatagramReader::new(),
            pending: std::collections::VecDeque::new(),
            relay_addr: Some(relay_addr),
        })
    }

    /// The adopted channel's id (observability + correlation; one
    /// session = one channel).
    pub fn channel_id(&self) -> u32 {
        self.channel_id
    }

    /// The reply's sentinel `BND.ADDR` (RFC fidelity: the relay
    /// address the vanilla shape would sendto; the wrapper-aware
    /// session never uses it — the handle is the channel, ADR-003).
    /// `None` after a taken stream (the spent-session posture).
    pub fn relay_addr(&self) -> Option<TargetAddr> {
        self.relay_addr.clone()
    }

    /// Send one datagram to `dst`: the RFC address encoding + the
    /// payload, length-framed **exactly once** by the wire codec
    /// ([`frame`] — the framing-once mirror, ADR-003) and written to
    /// the channel. A zero-length payload is a legal empty datagram
    /// (the F-2 invariant at payload level). Rejects
    /// [`SocksSessionError::Rfc`] ([`WireError`]) at frame time —
    /// the joint bound (`len(addr) + len(payload) > 65535`, the u16
    /// length field would wrap) and the address-encoding refusals.
    pub async fn send_datagram(
        &mut self,
        dst: impl ToTargetAddr,
        payload: &[u8],
    ) -> Result<(), SocksSessionError> {
        let dst: TargetAddr = dst.to_target_addr()?;
        let framed = frame(&dst, payload).map_err(wire_error)?;
        let Some(inner) = self.inner.as_mut() else {
            return Err(SocksSessionError::Io(std::io::Error::new(
                std::io::ErrorKind::NotConnected,
                "the association's stream is spent",
            )));
        };
        let socket = inner.get_socket_mut();
        socket.write_all(&framed).await?;
        socket.flush().await?;
        Ok(())
    }

    /// Receive the next datagram: decode incrementally from the
    /// channel's read half — `(source_addr, payload)` per datagram,
    /// `Ok(None)` on channel EOF (the association's end — the
    /// association lifetime is the channel's, ADR-003).
    ///
    /// **The reader state persists across calls** (POC #2's finding,
    /// ADR-003): the incremental [`DatagramReader`]'s mid-frame
    /// position survives a caller-driven receive loop — a datagram
    /// split across channel chunks reassembles across calls — and
    /// frames decoded beyond the one returned (a chunk may batch
    /// several datagrams) queue in `pending` for later calls.
    ///
    /// Fail-loud shapes (never a silent partial datagram): a stream
    /// end mid-frame is the truncated-datagram condition (an
    /// `UnexpectedEof` I/O error through [`SocksSessionError::Rfc`]);
    /// a frame the codec refuses (garbage ATYP — any post-reply
    /// control byte) errors loud through [`WireError`]'s
    /// `UdpHeaderError` mapping and ends the caller's loop.
    pub async fn recv_datagram(
        &mut self,
    ) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError> {
        // The batched-datagram pending queue first (a chunk may have
        // batched several frames — drain in order).
        if let Some(dg) = self.pending.pop_front() {
            return Ok(Some(dg));
        }

        let mut chunk = vec![0u8; 16 * 1024];
        let Some(inner) = self.inner.as_mut() else {
            return Ok(None);
        };
        let socket = inner.get_socket_mut();
        loop {
            let n = socket.read(&mut chunk).await?;
            if n == 0 {
                // Channel EOF = association end (the clean exit) —
                // fail-loud if a frame was in flight (never a silent
                // partial datagram; the reader state is the teardown
                // diagnostic).
                if self.reader.is_mid_frame() {
                    return Err(SocksSessionError::Rfc(SocksError::Io(std::io::Error::new(
                        std::io::ErrorKind::UnexpectedEof,
                        "association ended mid-datagram (fail-loud)",
                    ))));
                }
                return Ok(None);
            }
            let mut dgs = self.reader.feed(&chunk[..n]).into_iter();
            if let Some(first) = dgs.next() {
                for res in dgs {
                    match res {
                        Ok(pair) => self.pending.push_back(pair),
                        Err(err) => return Err(wire_error(err)),
                    }
                }
                match first {
                    Ok(pair) => return Ok(Some(pair)),
                    Err(err) => return Err(wire_error(err)),
                }
            }
        }
    }

    /// The underlying stream as split read/write halves (the front
    /// door's bridge seam: a raw pipe to a vanilla client over the
    /// framed stage — the door's RFC header codec owns the frames at
    /// the boundary, ADR-003). `None` after the stream was taken (the
    /// spent-session posture).
    pub fn split(&mut self) -> Option<SplitPipe<'_>> {
        let inner = self.inner.as_mut()?;
        Some(tokio::io::split(inner))
    }

    /// Take the underlying [`Socks5Stream<BiStream>`] (advanced use —
    /// the forked client type's socket accessors, owned-half pumps;
    /// the POC's `raw_mut` seam, widened). The session rides along
    /// still owning the channel's teardown: drop the pair (or
    /// `shutdown`/`close` the session) when done.
    ///
    /// [`SpentSessionError`] on a spent session (the stream already
    /// taken): a spent session's manager entry is always reaped
    /// already (`shutdown`/`close`/`Drop` all take it), so dropping it
    /// loses nothing — no panic path (AGENTS.md convention 2), and the
    /// `Err` side stays small (a spent session is not carried).
    pub fn into_inner(mut self) -> Result<TakenAssociateInner, SpentSessionError> {
        match self.inner.take() {
            Some(inner) => Ok(TakenAssociateInner {
                session: self,
                inner,
            }),
            None => Err(SpentSessionError),
        }
    }

    /// Half-close the stream (the association's graceful end): the
    /// mux's implicit-EOF sentinel reaches the producer's stage as
    /// EOF and the association ends (association lifetime = channel
    /// lifetime, ADR-003). Returns whether the adopted channel entry
    /// existed (the reap result; the session is spent either way).
    pub async fn shutdown(mut self) -> Result<bool, SocksSessionError> {
        let reaped = if let Some(inner) = self.inner.as_mut() {
            let socket = inner.get_socket_mut();
            socket.shutdown().await?;
            self.inner.take();
            self.reap()
        } else {
            self.reap()
        };
        Ok(reaped)
    }

    /// Tear the association down gracefully (the CONNECT session's
    /// `close` posture): half-close the stream (the EOF sentinel
    /// reaches the producer's stage), then reap the adopted channel
    /// entry. Returns whether the entry existed.
    pub async fn close(mut self) -> bool {
        if let Some(inner) = self.inner.as_mut() {
            let _ = inner.shutdown().await;
        }
        self.inner.take();
        self.reap()
    }

    /// The adopted channel's teardown (`teardown_channel` returns a
    /// producer-side pump-task handle the caller may abort after the
    /// drain — the channels-call path's shape). The handle is
    /// deliberately dropped here: the producer-side pump finishes via
    /// EOF propagation (the graceful shape — aborting would cut
    /// mid-flight data); `clear_all` aborts instead (the substrate's
    /// abrupt path).
    fn reap(&mut self) -> bool {
        self.manager
            .take()
            .map(|m| m.teardown_channel(self.channel_id).is_ok())
            .unwrap_or(false)
    }
}

impl Drop for AssociateSession {
    /// Sync-reap the adopted channel entry (`teardown_channel` is
    /// sync — best-effort; `Drop` cannot await) and drop the pipe
    /// (the mux's implicit-EOF sentinel EOFs the producer's stage —
    /// the association ends with the session's lifetime). Dropping
    /// without `shutdown`/`close` never leaks the channel entry.
    fn drop(&mut self) {
        self.inner.take();
        let _ = self.reap();
    }
}

/// The owned stream [`AssociateSession::into_inner`] yields, paired
/// with the session that still owns the channel's teardown.
pub struct TakenAssociateInner {
    /// The teardown owner — reap via `shutdown`/`close`/`Drop`.
    pub session: AssociateSession,
    /// The framed datagram stage's stream (the forked client type
    /// over the channel's `BiStream`).
    pub inner: Socks5Stream<BiStream>,
}

/// The owned pipe [`Socks5ChannelSession::into_inner`] yields, paired
/// with the session that still owns the channel's teardown.
pub struct TakenInner {
    /// The teardown owner — reap via `close`/`Drop`.
    pub session: Socks5ChannelSession,
    /// The post-CONNECT pipe (the forked client type over the
    /// channel's `BiStream`).
    pub inner: Socks5Stream<BiStream>,
}

/// The borrowed halves [`Socks5ChannelSession::split`] yields (the
/// `tokio::io::split` pair over the pipe). A half-close rides the
/// write half's [`AsyncWriteExt::shutdown`].
pub type SplitPipe<'s> = (
    ReadHalf<&'s mut Socks5Stream<BiStream>>,
    WriteHalf<&'s mut Socks5Stream<BiStream>>,
);
/// A completed BIND reply#1 (consumer.md §The BIND client, ADR-007's
/// client-side phase reading): the channel is open, reply#1 carried
/// BND.ADDR — the advertised address the app server dials (the one
/// real address on the channels path: the local variant's bound
/// listener; the composed variant's expected far-side address; the
/// wrapper-aware sentinel `0.0.0.0:0` on the in-band default). The
/// next phase is the session's [`BindSession::await_inbound`].
///
/// The handle is a **view**, not a reader (review 002 D-1's
/// one-reader decision): it carries the advertised address and the
/// raw pre-reply#2 stream ([`Self::raw_mut`], the front-door bridge
/// seam); the one normative reply#2 reader is
/// [`BindSession::await_inbound`] alone, take-guarded there — a
/// fresh handle per call cannot double-read anything.
///
/// The session owns the adopted channel — teardown stays there
/// (reap via the session; this handle borrows into it). No `Clone`
/// (the compile-assert rides [`BindSession`]).
pub struct BindReply1<'s> {
    inner: &'s mut Socks5Stream<BiStream>,
    /// Reply#1's BND.ADDR — the advertised address.
    pub bind_addr: TargetAddr,
}

impl BindReply1<'_> {
    /// The raw stream (the front-door bridge seam: a raw pipe to a
    /// vanilla client, pre-reply#2).
    pub fn raw_mut(&mut self) -> &mut Socks5Stream<BiStream> {
        self.inner
    }
}

/// The post-reply#2 data plane (the pass-through to the accepted
/// inbound peer): 0 B framing — reads/writes ride the channel
/// directly. The plane **owns** the stream (review 002 C-1's
/// take-shape — the taken inner pairs with the session that stays
/// the teardown owner, the `TakenInner` pairing posture); the
/// [`BindSession`] keeps the channel's teardown — drop it (or
/// `close` it) when done; dropping the plane EOFs the pipe
/// structurally (the mux's implicit-EOF sentinel). Accessors:
/// [`Self::into_inner`] (owned) / [`Self::get_socket_mut`]
/// (borrowed; the POC shape).
pub struct BindDataPlane {
    inner: Socks5Stream<BiStream>,
    /// Reply#2's peer address (the RFC's "the address of the incoming
    /// connection" — the attribution surface, in-band).
    pub peer: TargetAddr,
}

impl BindDataPlane {
    /// Take the underlying [`Socks5Stream<BiStream>`] out of the data
    /// plane (the owned pipe; the forked client type). The session
    /// rides along still owning the teardown (`BindSession`'s
    /// `close`/`Drop`).
    pub fn into_inner(self) -> Socks5Stream<BiStream> {
        self.inner
    }

    /// The underlying stream (the forked client type's socket
    /// accessor, borrowed — the POC's `get_socket_mut` shape).
    pub fn get_socket_mut(&mut self) -> &mut BiStream {
        self.inner.get_socket_mut()
    }
}

impl tokio::io::AsyncRead for BindDataPlane {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl tokio::io::AsyncWrite for BindDataPlane {
    fn poll_write(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        std::pin::Pin::new(&mut self.inner).poll_write(cx, buf)
    }

    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
    }

    fn poll_shutdown(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
    }
}

/// The BIND consumer session (the `bind()` entry namespace — the POC
/// #5 shape): opens the channel, negotiates, sends BIND, reads
/// reply#1 (`bind`), then awaits reply#2 + the data plane
/// (`await_inbound`). The session owns the adopted channel: on Drop
/// it reaps the entry (the CONNECT session's teardown posture). It
/// does **not** implement `Clone`:
///
/// ```compile_fail
/// use alksocks::consumer::BindSession;
/// fn assert_clone<T: Clone>() {}
/// assert_clone::<BindSession>();
/// ```
///
/// Error surface: a refused BIND (the command gate's 0x02) or a
/// factory failure's reply#1 code surfaces from `bind` as
/// [`SocksSessionError::Rfc`] (the adopted channel is torn down
/// before the error returns — teardown-at-birth, the phantom-channel
/// property); accept-stage failures surface at `await_inbound` as
/// reply#2's mapped code ([`SocksSessionError::Rfc`]) — the channel
/// is still open there (the listener existed; the session lives until
/// the app closes it; the inner is restored, see the take-guard
/// below).
///
/// The reply#2 phase is take-guarded (review 002
/// C-1): the first `await_inbound` takes the inner stream — the
/// returned plane owns it (the `TakenInner` pairing posture; the
/// session stays the teardown owner) — so a second call hits the
/// `None` arm and returns `Ok(None)`: the session family's spent
/// shape, with no re-read of the live data plane possible. A failed
/// read restores the inner (the phase is only consumed on success;
/// the producer ends the session on every post-reply#1 failure exit,
/// so a re-call EOFs — never a live data-plane re-read either way).
pub struct BindSession {
    channel_id: u32,
    manager: Option<ChannelManager>,
    inner: Option<Socks5Stream<BiStream>>,
    /// Reply#1's BND.ADDR — captured at `bind` (the forked client's
    /// `request()` returns it once; the session carries it for the
    /// app server's out-of-band dial).
    bind_addr: Option<TargetAddr>,
}

impl std::fmt::Debug for BindSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("BindSession")
            .field("channel_id", &self.channel_id)
            .field("awaiting_inbound", &self.inner.is_some())
            .field("bind_addr", &self.bind_addr)
            .finish()
    }
}

impl BindSession {
    /// BIND through a produced SOCKS5 resource: opens the channel
    /// (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)` — the trivial
    /// params, ADR-001), drives the no-auth handshake and the BIND
    /// request in-stream over the channel's `BiStream`, and reads
    /// reply#1 (carrying BND.ADDR — the forked client `request()`'s
    /// normal path). The returned session owns the channel; the
    /// advertised address is [`BindReply1::bind_addr`].
    ///
    /// Error surface: an open failure is [`SocksSessionError::Open`]
    /// (the typed ADR-049 reasons; no channel was ever returned); a
    /// refused BIND — the command gate's `ConnectionNotAllowed` 0x02
    /// or a factory failure's mapped reply#1 code — surfaces the RFC
    /// reply code through [`SocksSessionError::Rfc`], and the adopted
    /// channel is torn down before the error returns (no phantom
    /// channel).
    pub async fn bind(
        client: &ChannelClient,
        target: impl ToTargetAddr,
    ) -> Result<Self, SocksSessionError> {
        let target: TargetAddr = target.to_target_addr()?;
        let (channel_id, inner, manager) = open_and_negotiate(client, |mut inner| async move {
            let bind_addr = inner.request(Socks5Command::TCPBind, target).await?;
            Ok((bind_addr, inner))
        })
        .await?;
        let (bind_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            bind_addr: Some(bind_addr),
        })
    }

    /// The cred-bearing front door's BIND re-origination (the
    /// token-carrying twin of [`Self::bind`]): the open rides the
    /// payload `auth_token` (CF-005 precedence (1) — the producer's
    /// identity provider resolves the door's mapped identity; the
    /// BIND command gate reads `socks5:bind` from it, ADR-006), then
    /// the same in-stream BIND phase model over the channel's
    /// `BiStream`. Error surface identical.
    pub async fn bind_with_token(
        client: &TokenClient,
        target: impl ToTargetAddr,
        token: Vec<u8>,
    ) -> Result<Self, SocksSessionError> {
        let target: TargetAddr = target.to_target_addr()?;
        let (channel_id, inner, manager) =
            open_with_token_and_negotiate(client, token, |mut inner| async move {
                let bind_addr = inner.request(Socks5Command::TCPBind, target).await?;
                Ok((bind_addr, inner))
            })
            .await?;
        let (bind_addr, inner) = inner;

        Ok(Self {
            channel_id,
            manager: Some(manager),
            inner: Some(inner),
            bind_addr: Some(bind_addr),
        })
    }

    /// The adopted channel's id (observability + correlation; one
    /// session = one channel).
    pub fn channel_id(&self) -> u32 {
        self.channel_id
    }

    /// Reply#1's view: the advertised BND.ADDR (the app server dials
    /// it out-of-band — the RFC's original role for the address) and
    /// the raw pre-reply#2 stream ([`BindReply1::raw_mut`], the
    /// front-door bridge seam). A **view**, not a reader (review 002
    /// D-1's one-reader decision): the one normative reply#2 reader
    /// is [`Self::await_inbound`]; the handle can be re-borrowed
    /// freely — it carries no read phase.
    pub fn reply1(&mut self) -> Option<BindReply1<'_>> {
        let inner = self.inner.as_mut()?;
        Some(BindReply1 {
            inner,
            bind_addr: self.bind_addr.clone()?,
        })
    }

    /// Await the inbound accept (reply#2) and take the pass-through
    /// data plane. The one normative reply#2 reader (review 002 C-1 +
    /// D-1): the phase is **take-guarded** — the read consumes the
    /// session's inner stream (the plane owns it; the session stays
    /// the teardown owner) — so a second call cannot re-read the
    /// live data plane. Reply#2 is hand-read with the forked
    /// `read_address` codec — reply#1's codec applied again, no
    /// second codec (POC #5's `read_reply2` sketch; ADR-011's
    /// rationale).
    ///
    /// `Ok(None)` after the plane was taken (the spent-session
    /// posture — the doc claim this guard makes true). A failed read
    /// restores the inner (the phase is not consumed on failure): the
    /// channel stays owned (reply#1 succeeded — the listener existed;
    /// reap via `close`/`Drop`), the pipe stays reachable
    /// (`split`), and a re-call re-reads against the already-ended
    /// session (the producer ends it on every post-reply#1 failure
    /// exit — the read EOFs; never a live data-plane re-read).
    pub async fn await_inbound(&mut self) -> Result<Option<BindDataPlane>, SocksSessionError> {
        let Some(mut inner) = self.inner.take() else {
            return Ok(None);
        };
        match read_reply2(&mut inner).await {
            Ok(peer) => Ok(Some(BindDataPlane { inner, peer })),
            Err(err) => {
                self.inner = Some(inner);
                Err(err)
            }
        }
    }

    /// The pass-through pipe as split read/write halves (0 B framing
    /// — raw bytes to/from the accepted peer). Valid in the reply#1
    /// phase (pre-reply#2 bytes ride the same channel — the
    /// front-door bridge seam); post-[`Self::await_inbound`] the
    /// plane owns the stream (the take-guard), so the data plane's
    /// halves ride `BindDataPlane`/its `into_inner` instead. `None`
    /// after a graceful `close` (the pipe is dropped with the reap).
    pub fn split(&mut self) -> Option<SplitPipe<'_>> {
        let inner = self.inner.as_mut()?;
        Some(tokio::io::split(inner))
    }

    /// Tear the session down gracefully: half-close the pipe (the EOF
    /// sentinel reaches the producer's pump), then reap the adopted
    /// channel entry. Returns whether the entry existed.
    pub async fn close(mut self) -> bool {
        if let Some(inner) = self.inner.as_mut() {
            let _ = inner.shutdown().await;
        }
        self.inner.take();
        self.reap()
    }

    /// The adopted channel's teardown (`teardown_channel` returns a
    /// producer-side pump-task handle the caller may abort after the
    /// drain — the channels-call path's shape). The handle is
    /// deliberately dropped here: the producer-side pump finishes via
    /// EOF propagation (the graceful shape — aborting would cut
    /// mid-flight data); `clear_all` aborts instead (the substrate's
    /// abrupt path).
    fn reap(&mut self) -> bool {
        self.manager
            .take()
            .map(|m| m.teardown_channel(self.channel_id).is_ok())
            .unwrap_or(false)
    }
}

impl Drop for BindSession {
    /// Sync-reap the adopted channel entry (`teardown_channel` is
    /// sync — best-effort; `Drop` cannot await) and drop the pipe
    /// (the mux's implicit-EOF sentinel EOFs the producer's pump).
    /// Dropping without `close` never leaks the channel entry.
    fn drop(&mut self) {
        self.inner.take();
        let _ = self.reap();
    }
}

/// Map a datagram-frame decode error ([`WireError`]) into the RFC
/// vocabulary the session surface carries: the address-encoding
/// refusal is the forked [`AddrError`] class (the same sub-layout
/// the forked codec validates — one vocabulary), the protocol-garbage
/// shapes (bad ATYP / a short frame / non-UTF-8 domain / a `len == 0`
/// frame) surface as the RFC UDP-header error class
/// ([`UdpHeaderError`]) — the post-reply-garbage semantic, the
/// consumer-side mirror of the producer stage's
/// `UndecodableFrame` exit. The oversize family is unreachable on
/// the receive path (the length codec's frame-time bound makes a
/// declared length undecodable only via the garbage shapes) — it
/// maps through the UDP-header class too (fail-loud either way).
fn wire_error(err: WireError) -> SocksSessionError {
    match err {
        WireError::Address(addr) => SocksError::AddrError(addr).into(),
        other => SocksError::UdpHeaderError(UdpHeaderError::ReadingError(std::io::Error::other(
            other.to_string(),
        )))
        .into(),
    }
}

/// Read reply#2 off the raw stream (the RFC's second reply: VER, REP,
/// RSV, ATYP, addr, port — the POC #5 `read_reply2` sketch): the 4-byte
/// head, the version pin, the reply code mapped through the forked
/// `ReplyError::from_u8`, then the address via the forked public
/// `read_address` codec — reply#1's codec applied again, **no second
/// codec** (ADR-011's rationale).
async fn read_reply2<T>(inner: &mut Socks5Stream<T>) -> Result<TargetAddr, SocksSessionError>
where
    T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    let socket = inner.get_socket_mut();
    let mut header = [0u8; 4];
    socket.read_exact(&mut header).await?;
    let [version, reply, _rsv, atyp] = header;
    if version != consts::SOCKS5_VERSION {
        return Err(SocksError::UnsupportedSocksVersion(version).into());
    }
    if reply != consts::SOCKS5_REPLY_SUCCEEDED {
        return Err(SocksError::ReplyError(ReplyError::from_u8(reply)).into());
    }
    let peer = read_address(socket, atyp)
        .await
        .map_err(SocksError::AddrError)?;
    Ok(peer)
}

/// The cred-bearing front door's channel surface (the
/// token-carrying twin of a `ChannelClient`'s open path): wraps the
/// client's channel-0 [`CallConnection`] so an open call can carry the
/// payload `auth_token` (CF-005 precedence (1) — the producer-side
/// dispatch resolves the token through its
/// `ServingConfig.identity_provider` and runs the ACL gate against the
/// resolved identity, ADR-017 §7). The client's call connection is
/// taken once at construction (the client's own calling surface ends
/// — a door client is a door client); the manager stays shared for
/// `adopt_channel`/teardown.
///
/// The token is the credential table's *identity id* (the resolution
/// seam's vocabulary — the SOCKS password itself never rides it, and
/// the vanilla client never sees the identity). The door's producer
/// resolves `token → Identity` through its
/// `ServingConfig.identity_provider`; the identity's scopes govern the
/// open gate and the BIND command gate exactly as any per-call opener
/// identity does (CF-005/CF-006).
pub struct TokenClient {
    call: Arc<CallConnection>,
    manager: ChannelManager,
}

impl TokenClient {
    /// Wrap a `ChannelClient`: take its call connection (the open-op
    /// calling surface) and clone the manager. The door's
    /// token-bearing sessions then ride this client.
    ///
    /// [`TokenClientError::CallConnectionTaken`] when the client's
    /// call connection was already taken (a client is a single
    /// calling surface — the door-only contract; convention 2: no
    /// panic path on a double wrap).
    pub async fn wrap(client: &ChannelClient) -> Result<Self, TokenClientError> {
        let call = client
            .take_call_connection()
            .await
            .ok_or(TokenClientError::CallConnectionTaken)?;
        Ok(Self {
            call,
            manager: client.manager().clone(),
        })
    }

    /// The manager (adopt/teardown — the token sessions' prologue uses
    /// it; the sessions' teardown owns it per session).
    pub fn manager(&self) -> &ChannelManager {
        &self.manager
    }

    /// The raw open call with the `auth_token` payload (CF-005
    /// precedence (1)): the payload shape mirrors
    /// [`ChannelClient::call_open_op`]'s `call.requested` envelope with
    /// the token added (ADR-017 §7 — the hub-forwarding path's shape).
    /// Returns the adopted `channel_id` (the reply's `channel_id`
    /// extracted — the adopt is the caller's next step, mirroring
    /// `open_channel`'s two halves).
    pub async fn open_with_token(&self, token: Vec<u8>) -> Result<u32, SocksOpenError> {
        let payload = serde_json::json!({
            "operationId": OP_SOCKS5_OPEN,
            "input": Socks5Params::as_json(),
            "auth_token": String::from_utf8_lossy(&token).into_owned(),
        });
        let response = self.call.call_with_payload(payload).await;
        let out = response
            .result
            .map_err(|error| ChannelOpenError::CallFailed { error })
            .map_err(SocksOpenError::from)?;
        let channel_id = out
            .get("channel_id")
            .and_then(|v: &serde_json::Value| v.as_u64())
            .ok_or(ChannelOpenError::MissingChannelId)? as u32;
        Ok(channel_id)
    }
}

/// Why a `TokenClient` refused to construct (the typed construction
/// error — the client's calling surface was already taken;
/// convention 2: no panic path).
#[derive(Debug, thiserror::Error)]
pub enum TokenClientError {
    #[error("the channel client's call connection was already taken (a client is one calling surface — the door-only contract)")]
    CallConnectionTaken,
}

impl std::fmt::Debug for TokenClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TokenClient").finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use alkcall::channels::client::ChannelOpenError;
    use futures::future::FutureExt as _;

    fn assert_send<T: Send>() {}

    #[test]
    fn session_shape_is_send_and_not_clone() {
        assert_send::<Socks5ChannelSession>();
        assert_send::<TakenInner>();
    }

    #[test]
    fn bind_session_shape_is_send_and_not_clone() {
        assert_send::<BindSession>();
        assert_send::<BindDataPlane>();
    }

    #[tokio::test]
    async fn spent_bind_await_inbound_returns_the_spent_shape() {
        // The C-1 take-guard's spent arm: a session whose inner is
        // gone (taken at the first read / the teardown) returns the
        // family's `Ok(None)` — no re-read, no panic.
        let mut session = BindSession {
            channel_id: 9,
            manager: None,
            inner: None,
            bind_addr: None,
        };
        let spent = session.await_inbound().await;
        assert!(
            matches!(spent, Ok(None)),
            "the spent shape: {:?}",
            spent.as_ref().err().map(|e| e.to_string())
        );
        assert!(session.reply1().is_none(), "no reply#1 view either");
        assert!(session.split().is_none(), "no pipe after teardown");
    }

    #[test]
    fn debug_impl_names_the_channel_and_liveness() {
        let session = Socks5ChannelSession {
            channel_id: 7,
            manager: None,
            inner: None,
            reply_addr: None,
        };
        let rendered = format!("{session:?}");
        assert!(rendered.contains("channel_id: 7"), "{rendered}");
        assert!(rendered.contains("connected: false"), "{rendered}");
    }

    #[test]
    fn spent_session_reap_is_a_no_op() {
        let mut session = Socks5ChannelSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reply_addr: None,
        };
        assert!(!session.reap(), "no manager — nothing reaped");
        assert!(session.split().is_none(), "no pipe after teardown");
    }

    #[test]
    fn spent_session_into_inner_is_an_err_not_a_panic() {
        // The `expect` the review removed (R-2): the spent-session
        // path is reachable (a session whose inner is gone), so it
        // must be a typed `Err` — no panic path.
        let session = Socks5ChannelSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reply_addr: None,
        };
        assert!(
            session.into_inner().is_err(),
            "a spent session must return Err, not Ok"
        );
    }

    #[test]
    fn open_errors_carry_the_typed_surface_through_connects_mapping() {
        // The mapping `connect` applies to an open failure: the typed
        // `SocksOpenError` vocabulary rides `SocksSessionError::Open`.
        let open = SocksOpenError::Open(ChannelOpenError::MissingChannelId);
        let session: SocksSessionError = open.into();
        assert!(matches!(session, SocksSessionError::Open(_)));
    }

    #[test]
    fn associate_session_shape_is_send_and_not_clone() {
        assert_send::<AssociateSession>();
        assert_send::<TakenAssociateInner>();
    }

    #[test]
    fn associate_debug_impl_names_the_channel_and_liveness() {
        let session = AssociateSession {
            channel_id: 7,
            manager: None,
            inner: None,
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        let rendered = format!("{session:?}");
        assert!(rendered.contains("channel_id: 7"), "{rendered}");
        assert!(rendered.contains("associated: false"), "{rendered}");
    }

    #[test]
    fn spent_associate_reap_is_a_no_op() {
        let mut session = AssociateSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        assert!(!session.reap(), "no manager — nothing reaped");
        assert!(session.split().is_none(), "no stream after teardown");
        assert!(session.relay_addr().is_none());
    }

    #[test]
    fn spent_associate_into_inner_is_an_err_not_a_panic() {
        // The `expect` the review removed (R-2): the spent-session
        // path is reachable (`into_inner` twice), so it must be a
        // typed `Err` — no panic path. The teardown owner rides back
        // to the caller either way.
        let session = AssociateSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        assert!(
            session.into_inner().is_err(),
            "a spent session must return Err, not Ok"
        );
    }

    #[tokio::test]
    async fn recv_datagram_persists_reader_state_across_calls() {
        use crate::wire::frame as wire_frame;
        use tokio::io::AsyncWriteExt as _;

        // The frame-source side: two frames, the first one split
        // across two writes (the 2-byte-prefix chunk shape — the POC
        // finding's trigger) and the second batched into the same
        // read batch. The session's reader state must survive the
        // call boundary: the first call returns the first datagram
        // (reassembled across reads), the second drains the pending
        // queue.
        let (mut writer, reader) = tokio::io::duplex(1024);
        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
        let wire = {
            let mut v = wire_frame(&addr, b"0123456789").unwrap().to_vec();
            v.extend_from_slice(&wire_frame(&addr, b"second").unwrap());
            v
        };
        writer.write_all(&wire[..5]).await.unwrap();
        writer.write_all(&wire[5..]).await.unwrap();
        drop(writer);

        let mut session = build_test_session(reader);

        let (src, first) = session
            .recv_datagram()
            .await
            .expect("recv")
            .expect("a datagram");
        assert_eq!(src, addr);
        assert_eq!(&first[..], b"0123456789");

        // The second frame was batched in the same read batch: the
        // pending queue carries it to the next call.
        let (src, second) = session
            .recv_datagram()
            .await
            .expect("recv")
            .expect("a datagram");
        assert_eq!(src, addr);
        assert_eq!(&second[..], b"second");

        // EOF: both frames consumed, the queue empty — `None`.
        assert!(session.recv_datagram().await.expect("eof").is_none());
    }

    #[tokio::test]
    async fn recv_datagram_fails_loud_when_stream_ends_mid_frame() {
        use tokio::io::AsyncWriteExt as _;

        let (mut writer, reader) = tokio::io::duplex(1024);
        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
        let framed = crate::wire::frame(&addr, b"payload").unwrap();
        writer.write_all(&framed[..4]).await.unwrap();
        drop(writer);

        let mut session = build_test_session(reader);
        let err = session
            .recv_datagram()
            .await
            .expect_err("a stream end mid-frame is the truncated-datagram condition");
        assert!(
            matches!(err, SocksSessionError::Rfc(SocksError::Io(_))),
            "fail-loud, not a silent partial: {err:?}"
        );
    }

    #[tokio::test]
    async fn recv_datagram_fails_loud_on_an_undecodable_frame() {
        use tokio::io::AsyncWriteExt as _;

        let (mut writer, reader) = tokio::io::duplex(1024);
        // Hand-craft a frame whose inner bytes start with an
        // unassigned ATYP (the post-reply-garbage semantic).
        let inner_bytes = vec![0x7Fu8, 1, 2, 3];
        let mut garbage = (inner_bytes.len() as u16).to_be_bytes().to_vec();
        garbage.extend_from_slice(&inner_bytes);
        writer.write_all(&garbage).await.unwrap();
        drop(writer);

        let mut session = build_test_session(reader);
        let err = session
            .recv_datagram()
            .await
            .expect_err("garbage ATYP is loud, never a silent drop");
        assert!(matches!(err, SocksSessionError::Rfc(_)), "{err:?}");
    }

    #[test]
    fn wire_error_maps_the_decode_garbage_shapes() {
        // The address-encoding refusal rides the forked AddrError
        // class; the garbage shapes ride the UDP-header class.
        let addr_err =
            crate::wire::frame(&TargetAddr::Domain("a".repeat(256), 80), b"").unwrap_err();
        let mapped = wire_error(addr_err);
        assert!(matches!(
            mapped,
            SocksSessionError::Rfc(SocksError::AddrError(_))
        ));

        let garbage = crate::wire::DatagramReader::new()
            .feed(&[0x00, 0x04, 0x7f, 1, 2, 3])
            .into_iter()
            .next()
            .unwrap()
            .unwrap_err();
        let mapped = wire_error(garbage);
        assert!(matches!(
            mapped,
            SocksSessionError::Rfc(SocksError::UdpHeaderError(_))
        ));
    }

    #[tokio::test]
    async fn send_datagram_on_a_spent_stream_is_the_io_arm() {
        // The spent-stream `Io` arm (review 002 N-8): a session whose
        // stream is gone errors loud through the typed I/O surface —
        // the mirror of the covered spent-`recv` arm.
        let mut session = AssociateSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        let err = session
            .send_datagram(TargetAddr::Ip("10.0.0.1:53".parse().unwrap()), b"payload")
            .await
            .expect_err("a spent stream must not send");
        assert!(
            matches!(err, SocksSessionError::Io(_)),
            "the spent-stream send is the Io arm: {err:?}"
        );

        // The spent-session read shape rides the same posture: the
        // `recv_datagram` spent arm is `Ok(None)` (the EOF-shape, the
        // family's read-side posture).
        assert!(
            session.recv_datagram().await.expect("spent recv").is_none(),
            "a spent stream reads the spent shape, never an error"
        );

        // The spent-session teardown surfaces: `shutdown` on a spent
        // session is a no-op reap (no manager — nothing reaped).
        let shutdown = session.shutdown().await.expect("spent shutdown");
        assert!(!shutdown, "no manager — nothing reaped");
    }

    #[tokio::test]
    async fn spent_associate_close_reaps_a_no_op() {
        // The `close`-on-spent arm: the graceful teardown path is a
        // no-op reap on a spent session (the session family's Drop
        // posture, on the async path).
        let session = AssociateSession {
            channel_id: 9,
            manager: None,
            inner: None,
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        assert!(!session.close().await, "no manager — nothing reaped");
    }

    #[tokio::test]
    async fn recv_datagram_mid_batch_later_frame_err_drops_the_pending_oks() {
        use tokio::io::AsyncWriteExt as _;

        // The mid-batch later-frame `Err` arm (review 002 N-8): the
        // first frame of a batch decodes `Ok`, a later one is garbage
        // — the batch is discarded with the session (fail-loud, never
        // a partial batch delivered while the session stays open).
        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
        let mut wire = crate::wire::frame(&addr, b"first").unwrap().to_vec();
        wire.extend_from_slice(&[0x00, 0x04, 0x7f, 1, 2, 3]);
        peer_writer.write_all(&wire).await.unwrap();
        drop(peer_writer);

        let (write_peer, session_writer) = tokio::io::duplex(64);
        let _ = write_peer;

        let mut session = AssociateSession {
            channel_id: 1,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        let err = session
            .recv_datagram()
            .await
            .expect_err("the later frame's garbage fails loud");
        assert!(
            matches!(err, SocksSessionError::Rfc(SocksError::UdpHeaderError(_))),
            "the decode garbage rides the UDP-header class: {err:?}"
        );

        // The pending `Ok` was dropped with the session — the queue
        // stays empty and the stream is spent for further reads (the
        // error ended the caller's loop, the fail-loud posture).
        assert!(session.pending.is_empty());
    }

    #[tokio::test]
    async fn associate_into_inner_pairs_the_stream_with_the_teardown_owner() {
        // The `Ok` arm of `AssociateSession::into_inner` (review 002
        // N-8): the taken stream pairs with the session that stays the
        // teardown owner (the `into_inner_pairs_the_pipe_with_the_
        // teardown_owner` shape's ASSOCIATE mirror) — and a second
        // take is the spent shape.
        let (_write_peer, session_writer) = tokio::io::duplex(64);
        let (read_peer, session_reader) = tokio::io::duplex(64);
        let _ = read_peer;

        let session = AssociateSession {
            channel_id: 1,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())),
        };

        let taken = session.into_inner().expect("the stream is live");
        // The pair carries both halves: the teardown owner (the relay
        // address still observable) and the owned stream.
        assert!(taken.session.relay_addr().is_some(), "the owner");
        assert!(taken.session.channel_id() == 1);
        drop(taken);
    }

    #[tokio::test]
    async fn associate_debug_renders_channel_liveness_and_pending() {
        // The Debug impl body (review 002 N-8, egress.rs class): the
        // associate session renders its channel id, liveness, the
        // relay address, and the pending count.
        let mut session = AssociateSession {
            channel_id: 7,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(
                tokio::io::duplex(16).0,
                tokio::io::duplex(16).0,
            )),
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())),
        };
        session.pending.push_back((
            TargetAddr::Ip("10.0.0.1:53".parse().unwrap()),
            Bytes::from_static(b"x"),
        ));
        let rendered = format!("{session:?}");
        assert!(rendered.contains("channel_id: 7"), "{rendered}");
        assert!(rendered.contains("associated: true"), "{rendered}");
        assert!(rendered.contains("relay_addr"), "{rendered}");
        assert!(rendered.contains("pending_datagrams: 1"), "{rendered}");
        drop(session);

        let bind = BindSession {
            channel_id: 8,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(
                tokio::io::duplex(16).0,
                tokio::io::duplex(16).0,
            )),
            bind_addr: Some(TargetAddr::Domain("bind.test".into(), 2121)),
        };
        let rendered = format!("{bind:?}");
        assert!(rendered.contains("channel_id: 8"), "{rendered}");
        assert!(rendered.contains("awaiting_inbound: true"), "{rendered}");
        assert!(rendered.contains("bind_addr"), "{rendered}");
    }

    #[tokio::test]
    async fn associate_split_yields_live_halves_and_survives_the_round_trip() {
        use crate::wire::frame as wire_frame;
        use tokio::io::AsyncWriteExt as _;

        // The front-door-bridge surface (consumer.md) on a *live*
        // stream: the halves borrow the session and the raw framed
        // bytes pass through both directions (0 B framing — the door's
        // RFC header codec would own the frames at the boundary).
        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
        let framed = wire_frame(&addr, b"split").unwrap().to_vec();

        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
        peer_writer.write_all(&framed).await.unwrap();
        drop(peer_writer);
        let (mut peer_reader, session_writer) = tokio::io::duplex(1024);

        let mut session = AssociateSession {
            channel_id: 1,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        };
        let (mut read, mut write) = session.split().expect("live halves");

        // Read the raw frame through the borrowed read half.
        let mut header = [0u8; 2];
        read.read_exact(&mut header)
            .await
            .expect("the frame header");
        let len = u16::from_be_bytes(header) as usize;
        let mut rest = vec![0u8; len];
        read.read_exact(&mut rest).await.expect("the frame body");
        assert_eq!(len, 7 + 5, "one v4-address datagram frame");
        drop(read);

        // The write half passes through (the bridge seam's shape) —
        // the peer receives the exact bytes.
        write.write_all(&framed).await.unwrap();
        write.flush().await.unwrap();
        drop(write);

        let mut got = vec![0u8; framed.len()];
        peer_reader
            .read_exact(&mut got)
            .await
            .expect("the peer read");
        assert_eq!(got, framed, "the write half rode the raw pipe");
    }

    /// An inner `Socks5Stream` over a true `BiStream` (the session/
    /// plane field type) for the BIND-side surfaces: reads come from
    /// `session_end` (the peer's writes), writes go out via the duplex
    /// whose other end the test holds. Two joined halves, the same
    /// shape `open_and_negotiate` assembles.
    fn build_skip_auth_bi_stream(
        session_end: tokio::io::DuplexStream,
        writer_end: tokio::io::DuplexStream,
    ) -> Socks5Stream<BiStream> {
        let mut config = Config::default();
        config.set_skip_auth(true);
        Socks5Stream::use_stream(BiStream::from_joined(session_end, writer_end), None, config)
            .now_or_never()
            .expect("skip_auth's use_stream is instantly ready")
            .expect("skip_auth's use_stream succeeds")
    }

    #[tokio::test]
    async fn bind_split_yields_live_halves_pre_reply2() {
        use tokio::io::AsyncWriteExt as _;

        // The BIND front-door-bridge surface (consumer.md): the split
        // halves in the reply#1 phase — pre-reply#2 bytes ride the
        // same channel (the doc's claim, exercised on a live stream).
        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
        peer_writer.write_all(b"raw").await.unwrap();
        drop(peer_writer);

        let (mut peer_reader, session_writer) = tokio::io::duplex(1024);
        let mut session = BindSession {
            channel_id: 1,
            manager: None,
            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
            bind_addr: Some(TargetAddr::Ip("127.0.0.1:2121".parse().unwrap())),
        };
        {
            let (mut read, mut write) = session.split().expect("live pre-reply2 halves");
            write.write_all(b"ack").await.unwrap();
            write.flush().await.expect("flush");
            let mut buf = [0u8; 3];
            read.read_exact(&mut buf).await.expect("the raw bytes");
            assert_eq!(&buf, b"raw");
        }

        // The write half reached the peer (the bridge seam's shape).
        let mut buf = [0u8; 3];
        peer_reader.read_exact(&mut buf).await.expect("the ack");
        assert_eq!(&buf, b"ack");
        drop(peer_reader);

        // The take-guard is untouched: the reader still reads the
        // reply#2 phase (the split halves were a view, not a
        // consumer) — the dropped peer EOFs it loud, never the spent
        // shape.
        let outcome = session.await_inbound().await;
        assert!(
            !matches!(outcome, Ok(None)),
            "the reply#2 phase was not consumed by the split"
        );
        session.close().await;
    }

    #[tokio::test]
    async fn bind_data_plane_poll_flush_and_poll_shutdown_pass_through() {
        use tokio::io::AsyncWriteExt as _;

        // The BIND door's half-close rides the plane's poll_shutdown
        // passthrough (and poll_flush rides the write path): write →
        // flush → shutdown — the bytes reach the peer and the EOF
        // sentinel follows.
        let (_session_writer_unused, session_reader) = tokio::io::duplex(64);
        let (mut peer_reader, session_writer) = tokio::io::duplex(64);
        let mut plane = BindDataPlane {
            inner: build_skip_auth_bi_stream(session_reader, session_writer),
            peer: TargetAddr::Ip("10.0.0.7:44444".parse().unwrap()),
        };
        plane.write_all(b"half").await.expect("write");
        plane.flush().await.expect("the poll_flush passthrough");
        plane
            .shutdown()
            .await
            .expect("the poll_shutdown passthrough");
        drop(plane);

        let mut seen = Vec::new();
        let mut buf = [0u8; 4];
        loop {
            match peer_reader.read(&mut buf).await {
                Ok(0) | Err(_) => break,
                Ok(n) => seen.extend_from_slice(&buf[..n]),
            }
        }
        assert_eq!(&seen, b"half", "the write rode through, then EOF");
    }

    #[tokio::test]
    async fn read_reply2_rejects_a_foreign_producers_version() {
        use tokio::io::AsyncWriteExt as _;

        // A foreign producer's reply#2 with a non-5 version: the
        // version pin is loud (never an aliased peer).
        let (mut peer_writer, session_reader) = tokio::io::duplex(64);
        peer_writer
            .write_all(&[0x04, 0x00, 0x00, 0x01, 10, 0, 0, 1, 0, 80])
            .await
            .unwrap();
        drop(peer_writer);

        let (_write_peer_unused, session_writer) = tokio::io::duplex(64);
        let mut inner = build_skip_auth_bi_stream(session_reader, session_writer);
        let err = read_reply2(&mut inner)
            .await
            .expect_err("a foreign version is refused");
        assert!(
            matches!(
                err,
                SocksSessionError::Rfc(SocksError::UnsupportedSocksVersion(0x04))
            ),
            "the version pin is loud: {err:?}"
        );
    }

    /// A session over a raw duplex (the reader internals' test
    /// harness — the handshake-free path: `skip_auth` bypasses the
    /// greeting, no reply is expected, the datagram stage rides
    /// raw).
    fn build_test_session(session_end: tokio::io::DuplexStream) -> AssociateSession {
        let mut config = Config::default();
        config.set_skip_auth(true);
        let (read, write) = tokio::io::split(session_end);
        let inner = Socks5Stream::use_stream(BiStream::from_joined(read, write), None, config)
            .now_or_never()
            .expect("skip_auth's use_stream is instantly ready")
            .expect("skip_auth's use_stream succeeds");
        AssociateSession {
            channel_id: 1,
            manager: None,
            inner: Some(inner),
            reader: DatagramReader::new(),
            pending: Default::default(),
            relay_addr: None,
        }
    }
}