consumer: BindSession — the reply#2 reader + e2e suite

- BindSession::bind (open → no-auth → BIND → reply#1, capturing
  BND.ADDR) + reply1() (the BindReply1 handle) + await_inbound (the
  POC #5 read_reply2 — hand-read with the forked read_address codec,
  no second codec, ADR-011) returning BindDataPlane (into_inner /
  get_socket_mut + AsyncRead/AsyncWrite forwards)
- Teardown ownership (the CONNECT session's posture): Drop sync-reap +
  structural EOF, close() graceful, teardown-at-birth on in-stream
  refusals, no Clone (compile_fail-asserted)
- Typed error surface: open failures ride SocksSessionError::Open
  (ADR-049 reasons); the command gate's 0x02 and factory-failure
  reply#1 codes surface from bind as Rfc; accept-stage reply#2 codes
  surface at await_inbound
- tests/consumer-bind.rs: 9 e2e tests over the harness (round-trip,
  data-plane accessors, gate refusal, both failure mappings,
  quiet-wait garbage, drop teardown + health, the CONNECT+BIND
  regression guard, the Send probe)

Verification: cargo test (259 incl. the doctest), clippy -D warnings
(native + wasm), fmt, both wasm checks, cargo doc clean.

src/consumer.rs

@@ -8,9 +8,12 @@
 //! `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). The ASSOCIATE and BIND
-//! sessions land with their tasks (`core-consumer-associate` /
-//! `core-consumer-bind`).
+//! 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. The ASSOCIATE session lands with
+//! its task (`core-consumer-associate`).
 //!
 //! Target construction rides [`ToTargetAddr`]: IP-string forms parse
 //! to IP targets; the domain form passes through unresolved (ADR-012
@@ -35,13 +38,14 @@
 use alkcall::channels::client::ChannelClient;
 use alkcall::channels::manager::ChannelManager;
 use alkcall::core::types::BiStream;
-use tokio::io::{AsyncWriteExt, ReadHalf, WriteHalf};
+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::util::target_addr::{TargetAddr, ToTargetAddr};
-use crate::rfc1928::Socks5Command;
+use crate::rfc1928::consts;
+use crate::rfc1928::util::target_addr::{read_address, TargetAddr, ToTargetAddr};
+use crate::rfc1928::{ReplyError, Socks5Command, SocksError};
 
 /// A completed SOCKS5 CONNECT session on the consumer side: the
 /// adopted channel + the post-reply byte pipe to the target.
@@ -191,6 +195,301 @@ pub type SplitPipe<'s> = (
     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 [`Self::await_inbound`].
+///
+/// The session owns the adopted channel — teardown stays here
+/// (reap via the paired `BindSession`; 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<'s> BindReply1<'s> {
+    /// Await reply#2 (the inbound accept) and take the pass-through
+    /// 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). A non-success
+    /// reply#2 errors with the mapped RFC reply code (accept-stage
+    /// failures surface here — no data plane).
+    pub async fn await_inbound(self) -> Result<BindDataPlane<'s>, SocksSessionError> {
+        let peer = read_reply2(self.inner).await?;
+        Ok(BindDataPlane {
+            inner: self.inner,
+            peer,
+        })
+    }
+
+    /// 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 `BindDataPlane`-shaped pass
+/// through): the proxied connection to the accepted inbound peer, 0 B
+/// framing — reads/writes ride the channel directly. Accessors:
+/// [`Self::into_inner`] (owned) / [`Self::get_socket_mut`] (borrowed;
+/// the POC shape). The teardown owner is the [`BindSession`] the
+/// `into_inner`/`get_socket_mut` borrows came from — drop it (or
+/// `close` it) when done; dropping the data plane EOFs the pipe
+/// structurally.
+pub struct BindDataPlane<'s> {
+    inner: &'s mut 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<'s> BindDataPlane<'s> {
+    /// 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) -> &'s mut 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<()>> {
+        let inner = &mut *self.inner;
+        std::pin::Pin::new(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>> {
+        let inner = &mut *self.inner;
+        std::pin::Pin::new(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<()>> {
+        let inner = &mut *self.inner;
+        std::pin::Pin::new(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<()>> {
+        let inner = &mut *self.inner;
+        std::pin::Pin::new(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).
+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, 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 = async {
+            let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?;
+            let bind_addr = inner.request(Socks5Command::TCPBind, target).await?;
+            Result::<(TargetAddr, Socks5Stream<BiStream>), SocksSessionError>::Ok((
+                bind_addr, inner,
+            ))
+        };
+
+        match negotiated.await {
+            Ok((bind_addr, inner)) => Ok(Self {
+                channel_id,
+                manager: Some(client.manager().clone()),
+                inner: Some(inner),
+                bind_addr: Some(bind_addr),
+            }),
+            Err(err) => {
+                let _ = client.manager().teardown_channel(channel_id);
+                Err(err)
+            }
+        }
+    }
+
+    /// 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). The
+    /// reply#2 reader is [`BindReply1::await_inbound`]; the returned
+    /// handle borrows the session (teardown stays owned).
+    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 — [`BindReply1::await_inbound`] reached through the
+    /// session (the borrow rides the session; teardown stays owned).
+    /// `None` after a taken plane (the spent-session posture).
+    pub async fn await_inbound(&mut self) -> Result<Option<BindDataPlane<'_>>, SocksSessionError> {
+        let Some(inner) = self.inner.as_mut() else {
+            return Ok(None);
+        };
+        let peer = read_reply2(inner).await?;
+        Ok(Some(BindDataPlane { inner, peer }))
+    }
+
+    /// The pass-through pipe as split read/write halves (0 B framing
+    /// — raw bytes to/from the accepted peer). Valid both in the
+    /// reply#1 phase (pre-reply#2 bytes ride the same channel — the
+    /// front-door bridge seam) and post-[`Self::await_inbound`] (the
+    /// data plane proper). `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()
+    }
+
+    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();
+    }
+}
+
+/// 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)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/lib.rs

@@ -29,7 +29,7 @@ pub mod params;
 pub mod producer;
 pub mod wire;
 
-pub use consumer::Socks5ChannelSession;
+pub use consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession};
 pub use egress::{
     egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError,
     BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle,

tasks/implementation/core-consumer-bind.md

@@ -1,7 +1,7 @@
 ---
 id: implementation/core-consumer-bind
 name: "Core — consumer BIND: BindSession (reply#2 reader) + e2e tests"
-status: pending
+status: completed
 depends_on: [implementation/core-consumer-connect, implementation/core-producer-bind]
 scope: narrow
 risk: medium
@@ -45,15 +45,15 @@ adapted to the crate's error surface):
 
 ## Acceptance Criteria
 
-- [ ] `BindSession` with `bind`/`await_inbound`/data-plane accessors
+- [x] `BindSession` with `bind`/`await_inbound`/data-plane accessors
       per the POC shape
-- [ ] Reply#2 read via the forked `read_address` codec (no second
+- [x] Reply#2 read via the forked `read_address` codec (no second
       codec)
-- [ ] Typed error surface (reply-code-carrying session errors)
-- [ ] Teardown ownership on Drop; no `Clone` (compile-asserted)
-- [ ] E2E tests through the harness (round-trip, gate refusal,
+- [x] Typed error surface (reply-code-carrying session errors)
+- [x] Teardown ownership on Drop; no `Clone` (compile-asserted)
+- [x] E2E tests through the harness (round-trip, gate refusal,
       failure mappings, regression guard)
-- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean
+- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean
 
 ## References
 
@@ -65,8 +65,74 @@ adapted to the crate's error surface):
 
 ## Notes
 
-> To be filled by implementation agent
+> Filled by the implementation agent (2026-09-24):
+
+- The POC's `BindReply1` handle is absorbed into the teardown-owning
+  session (the alktunnels ADR-005 posture the CONNECT session
+  pinned): `BindSession` captures reply#1's BND.ADDR at `bind` (the
+  forked client `request()` returns it once) and exposes the phase
+  through borrows — `reply1()` yields the `BindReply1` handle
+  (`bind_addr` + `raw_mut`), `await_inbound(&mut self)` reads reply#2
+  and returns the borrowed `BindDataPlane` (`Ok(None)` on a spent
+  session). The session stays the single teardown owner for every
+  shape; the POC's `(session, channel_id)` tuple and the
+  `BindDataPlane` ownership split collapse into one owner.
+- `read_reply2` is the POC's port: 4-byte head → version pin →
+  `ReplyError::from_u8` → the forked public `read_address` codec (no
+  second codec, ADR-011). Errors ride `SocksSessionError` (the
+  `SocksError` conversions routed through the forked vocabulary, per
+  the crate's error-surface convention — `AddrError`/`ReplyError` are
+  wrapped into `SocksError` first because the session error type
+  exposes `From<SocksError>`, not the RFC inner enums directly).
+- `BindDataPlane<'s>` borrows the session's stream (the CONNECT
+  session's `SplitPipe` shape, generalized): `into_inner()` yields
+  the `&mut Socks5Stream<BiStream>` (the teardown owner rides the
+  session), `get_socket_mut()` is the forked client's socket accessor
+  (the POC shape), and `AsyncRead`/`AsyncWrite` are forwarded so the
+  plane is a plain pipe (`pump_bidi`-shaped consumers work without
+  the accessors). `BiStream` is not `Sync`, so the plane pins `Send`
+  (runtime probe) — same shape the forked client type has.
+- Teardown: Drop = sync reap + structural EOF (the CONNECT posture);
+  `close()` = half-close (the EOF sentinel reaches the producer's
+  pump) then reap. Teardown-at-birth on the in-stream refusal path
+  (command-gate 0x02, factory-failure reply#1): the adopted channel
+  is reaped before the error returns. Post-reply#1 failures
+  (accept-stage, quiet-wait garbage) keep the channel — reply#1
+  succeeded, the listener existed; `close`/`Drop` reaps.
+- The quiet-wait garbage path surfaces at `await_inbound` as an
+  `Io`-shaped error (the producer ends the session with no reply#2;
+  the consumer's reply#2 read EOFs) — the client-side mirror of the
+  producer's `UnexpectedUdpControlGarbage` exit, pinned by test.
+- Harness reuse: no harness changes — the consumer-side topology's
+  `wire_consumer_side_with` + the producer-bind task's
+  `echo_accept`/`refusing_accept*` closures cover every test; the
+  parked-accept factory for the quiet-wait garbage test is
+  test-local (deterministic garbage-vs-accept branch).
 
 ## Summary
 
-> To be filled on completion
+Implemented the BIND consumer session in `src/consumer.rs` (POC #5's
+shape, adapted to the crate's teardown-owning session posture):
+`BindSession::bind` (open → no-auth → BIND → reply#1, capturing
+BND.ADDR), `reply1()` (the `BindReply1` handle — the app server's
+out-of-band dial address + `raw_mut`), `await_inbound` (the POC's
+`read_reply2` — hand-read with the forked `read_address` codec, no
+second codec) returning the `BindDataPlane` (`into_inner`/
+`get_socket_mut` + `AsyncRead`/`AsyncWrite` forwards), and the
+CONNECT session's teardown ownership (Drop sync-reap + structural
+EOF, `close()` graceful, teardown-at-birth on in-stream refusals, no
+`Clone` — `compile_fail`-asserted). Typed error surface throughout:
+open failures are `SocksSessionError::Open` (ADR-049 reasons); the
+command gate's 0x02 and factory-failure reply#1 codes surface from
+`bind` as `Rfc`; accept-stage reply#2 codes surface at
+`await_inbound`. E2E suite (`tests/consumer-bind.rs`, 9 tests over
+the harness): the round-trip (reply#1 BND.ADDR → reply#2 peer →
+pass-through echo both directions), the data-plane accessors
+(`into_inner`/`get_socket_mut`/`raw_mut` on one pipe), the
+command-gate refusal (typed 0x02, no phantom channel), the
+factory-failure (reply#1 code) and accept-stage (reply#2 code)
+mappings, quiet-wait garbage ending the session, drop teardown +
+connection health, the CONNECT+BIND regression guard on one
+registration, and the `Send` compile-shape probe. Verification:
+`cargo test` (259 incl. the doctest), clippy `-D warnings` (all
+targets), fmt, and both wasm checks clean.

tests/consumer-bind.rs

@@ -0,0 +1,489 @@
+//! The consumer BIND integration tests — `BindSession` end to end
+//! through the harness's consumer-side topology (the producer's real
+//! command-dispatch path over channels; the BIND phase model, ADR-007,
+//! driven by the wrapper-aware client, POC #5's `read_reply2` shape).
+//!
+//! Covered: bind → reply#1 (BND.ADDR as the harness accept advertises)
+//! → reply#2 (the peer address) → pass-through echo both directions;
+//! the command-gate refusal surfaced as a typed error (a
+//! `socks5:open`-only identity cannot bind); factory-failure and
+//! accept-stage-failure mappings surfaced; quiet-wait garbage from the
+//! client side ends the session; the three-command regression guard
+//! from the consumer side.
+
+mod harness;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use alksocks::consumer::{BindDataPlane, BindReply1, BindSession, Socks5ChannelSession};
+use alksocks::egress::{BindAcceptError, BindAcceptFactoryError};
+use alksocks::error::SocksSessionError;
+use alksocks::rfc1928::consts;
+use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
+use alksocks::SocksError as SocksErrorKind;
+
+use harness::{
+    echo_egress, refusing_accept, refusing_accept_factory, wire_consumer_side_with, ECHO_BIND_PORT,
+};
+
+/// Both scopes: the open gate (`socks5:open`) AND the command gate
+/// (`socks5:bind`) — a BIND session needs both grants explicitly
+/// (ADR-006: never implied).
+fn bind_scoped() -> alkcall::core::auth::Identity {
+    harness::both_scopes_identity()
+}
+
+/// The no-leaked-channels probe (the consumer-connect suite's shape,
+/// re-pinned for the BIND sessions): both managers hold only channel
+/// 0 — the producer's entry reaps by its open-op wrapper; the probe
+/// polls past the reap race.
+async fn no_leaked_channels(topo: &harness::ConsumerSideTopology) {
+    let deadline = std::time::Duration::from_secs(5);
+    let consumer_ids = topo.consumer.manager().channel_ids();
+    assert!(
+        consumer_ids.iter().all(|&id| id == 0),
+        "consumer leaked channel entries: {consumer_ids:?}"
+    );
+    tokio::time::timeout(deadline, async {
+        loop {
+            let producer_ids = topo.producer_manager.channel_ids();
+            if producer_ids.iter().all(|&id| id == 0) {
+                return;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("the producer's handler-exit teardown reaps the entry");
+}
+
+/// A session error that carries an RFC reply code (the typed
+/// reply-code-carrying shape the BIND paths promise).
+fn reply_code_of(err: &SocksSessionError) -> Option<u8> {
+    match err {
+        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => Some(code.as_u8()),
+        _ => None,
+    }
+}
+
+/// The happy path: `bind` → reply#1 carries the harness accept's
+/// advertised BND.ADDR (`ECHO_BIND_PORT` — the reply#1 carrier); the
+/// app server's out-of-band dial shape is the `BindReply1` handle;
+/// `await_inbound` reads reply#2 (the peer address, 10.0.0.7:44444 —
+/// the harness's fixed stand-in); the data plane two-pumps raw bytes
+/// both directions; `close` reaps the adopted channel.
+#[tokio::test]
+async fn bind_round_trip_reply1_reply2_then_pass_through() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(harness::echo_accept()),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let mut session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session binds");
+
+    // Reply#1: the advertised BND.ADDR (the harness accept's static
+    // in-process address).
+    let channel_id = session.channel_id();
+    assert!(channel_id >= 1, "a producer-allocated id");
+    assert!(topo.consumer.manager().has_channel(channel_id));
+    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle");
+    assert_eq!(
+        bind_addr,
+        TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))),
+        "reply#1 carries the advertised BND.ADDR"
+    );
+
+    // Reply#2: the accepted peer (the harness's fixed stand-in), then
+    // the raw pass-through data plane.
+    let mut plane = session
+        .await_inbound()
+        .await
+        .expect("reply#2 read")
+        .expect("the plane exists before reply#2");
+    assert_eq!(
+        plane.peer,
+        TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 7], 44444))),
+        "reply#2 carries the accepted peer"
+    );
+
+    // Phase 2: raw pass-through, both directions.
+    plane.write_all(b"ping").await.expect("write");
+    plane.flush().await.expect("flush");
+    let mut buf = [0u8; 4];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        plane.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"ping");
+
+    // Graceful teardown: half-close (the EOF sentinel) + reap.
+    drop(plane);
+    assert!(session.close().await, "the adopted channel entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// The `BindDataPlane` accessors: `into_inner` yields the owned
+/// forked client type (read/write rides it directly); the session
+/// still owns the teardown — `close` after the plane is drained reaps
+/// the entry.
+#[tokio::test]
+async fn bind_data_plane_accessors_into_inner_and_socket_mut() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(harness::echo_accept()),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let mut session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("plane.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session binds");
+    session
+        .reply1()
+        .expect("reply#1's handle")
+        .raw_mut()
+        .get_socket_mut()
+        .write_all(b"pre")
+        .await
+        .expect("pre-reply#2 write");
+    let mut plane = session
+        .await_inbound()
+        .await
+        .expect("reply#2 read")
+        .expect("the plane");
+    // `get_socket_mut` is the borrowed socket accessor (the POC
+    // shape) — the same stream, reached through the plane.
+    plane
+        .get_socket_mut()
+        .write_all(b"raw")
+        .await
+        .expect("write");
+    let inner = plane.into_inner();
+    let mut buf = [0u8; 6];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        inner.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"preraw", "both write paths rode the same pipe");
+    let _unused = inner;
+
+    assert!(session.close().await, "the entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// The command gate (ADR-006): a `socks5:open`-only identity cannot
+/// bind — the refusal surfaces as a typed session error carrying the
+/// RFC reply code (`ConnectionNotAllowed` 0x02), and no phantom
+/// channel remains on either side.
+#[tokio::test]
+async fn bind_command_gate_refusal_is_a_typed_reply_code_error() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(harness::echo_accept()),
+        Some(harness::consumer_identity()),
+    )
+    .await;
+
+    let err = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("gate.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect_err("the gate refuses a socks5:open-only identity");
+
+    assert_eq!(
+        reply_code_of(&err),
+        Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED),
+        "the command gate's 0x02 rides the session error: {err:?}"
+    );
+
+    // Teardown-at-birth: the adopted channel was reaped before the
+    // error returned (no phantom channel, either side).
+    no_leaked_channels(&topo).await;
+}
+
+/// Factory failure (pre-reply#1) maps to reply#1's code, surfaced
+/// through the session type — no session lives past the refusal, no
+/// phantom listener.
+#[tokio::test]
+async fn bind_factory_failure_surfaces_reply1_code() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(refusing_accept_factory(BindAcceptFactoryError::Refused(
+            "port held".into(),
+        ))),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let err = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("factory.fail", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect_err("the factory refused");
+
+    assert_eq!(
+        reply_code_of(&err),
+        Some(consts::SOCKS5_REPLY_CONNECTION_REFUSED),
+        "the factory refusal maps to reply#1's code: {err:?}"
+    );
+    no_leaked_channels(&topo).await;
+}
+
+/// Accept-stage failure (post-reply#1) maps to reply#2's code —
+/// surfaced at `await_inbound`, no data plane; the session's channel
+/// is still owned (close/drop reaps it).
+#[tokio::test]
+async fn bind_accept_stage_failure_surfaces_reply2_code() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(refusing_accept(BindAcceptError::NotAllowed(
+            "policy".into(),
+        ))),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let mut session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("accept.fail", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("reply#1 succeeded — the factory resolved");
+
+    let err = match session.await_inbound().await {
+        Err(err) => err,
+        Ok(Some(_)) => panic!("the accept-stage refusal must not yield a data plane"),
+        Ok(None) => panic!("the plane existed before reply#2"),
+    };
+    assert_eq!(
+        reply_code_of(&err),
+        Some(consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED),
+        "reply#2's mapped code surfaces at await_inbound: {err:?}"
+    );
+
+    // The channel entry stays adopted (reply#1 succeeded; the session
+    // owns it until the app closes it) — close reaps it.
+    assert!(session.close().await, "the entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// Quiet-wait garbage from the client side: a control byte after
+/// reply#1 is protocol garbage — the producer ends the session with
+/// no data plane, so the consumer's `await_inbound` read EOFs (an I/O
+/// error, not a reply#2).
+#[tokio::test]
+async fn bind_quiet_wait_garbage_from_the_client_ends_the_session() {
+    let parked: alksocks::egress::BindAcceptFactory =
+        std::sync::Arc::new(move |_target: TargetAddr| {
+            Box::pin(async move {
+                Ok(alksocks::egress::BindAccept::new(
+                    std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)),
+                    std::future::pending::<
+                        Result<alksocks::egress::BindTargetHandle, BindAcceptError>,
+                    >(),
+                ))
+            })
+        });
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(parked),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let mut session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("quiet.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("reply#1 succeeded");
+
+    // Protocol garbage in the quiet wait (the client's side).
+    session
+        .reply1()
+        .expect("reply#1's handle")
+        .raw_mut()
+        .get_socket_mut()
+        .write_all(&[0x05, 0x02])
+        .await
+        .expect("garbage written");
+
+    // The session ends: the reply#2 read EOFs (no reply#2, no data
+    // plane).
+    let err = match session.await_inbound().await {
+        Err(err) => err,
+        Ok(Some(_)) => panic!("the garbage must not yield a data plane"),
+        Ok(None) => panic!("the plane existed before reply#2"),
+    };
+    assert!(
+        matches!(err, SocksSessionError::Io(_)),
+        "the quiet-wait garbage surfaces as the EOF'd reply#2 read: {err:?}"
+    );
+
+    // The channel is still owned (reply#1 succeeded); close reaps it.
+    assert!(session.close().await, "the entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// The three-command regression guard from the consumer side (POC #5
+/// test 7's consumer-side mirror): one registration, all three
+/// commands served through their consumer sessions — CONNECT
+/// (`Socks5ChannelSession`), BIND (`BindSession`), ASSOCIATE rides
+/// its task; here CONNECT + BIND coexist on one registration with the
+/// both-scopes identity, fresh channel per session.
+#[tokio::test]
+async fn bind_and_connect_serve_on_the_same_registration() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        Some(echo_egress()),
+        Some(harness::echo_accept()),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    // CONNECT serves (the CONNECT session over the same registration).
+    let mut connect_session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("echo.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the connect session");
+    connect_session
+        .split()
+        .expect("pipe")
+        .1
+        .write_all(b"tcp")
+        .await
+        .expect("write");
+    let (mut read, _) = connect_session.split().expect("pipe");
+    let mut buf = [0u8; 3];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        read.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"tcp");
+    connect_session.close().await;
+
+    // BIND serves the same registration (fresh channel).
+    let mut bind_session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the bind session");
+    let mut plane = bind_session
+        .await_inbound()
+        .await
+        .expect("reply#2")
+        .expect("the plane");
+    plane.write_all(b"bind").await.expect("write");
+    plane.flush().await.expect("flush");
+    let mut buf = [0u8; 4];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        plane.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"bind");
+    drop(plane);
+    bind_session.close().await;
+    no_leaked_channels(&topo).await;
+}
+
+/// Drop teardown: dropping a live BIND session mid-pipe reaps the
+/// adopted channel entry (no leak) and the connection stays healthy —
+/// a fresh bind succeeds immediately after (the CONNECT session's
+/// drop posture, re-pinned for BIND).
+#[tokio::test]
+async fn bind_drop_teardown_reaps_and_connection_stays_healthy() {
+    let topo = wire_consumer_side_with(
+        harness::echo_dial(),
+        None,
+        Some(harness::echo_accept()),
+        Some(bind_scoped()),
+    )
+    .await;
+
+    let mut session = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("dropped.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session binds");
+    let BindReply1 { bind_addr, .. } = session.reply1().expect("reply#1's handle");
+    assert_eq!(
+        bind_addr,
+        TargetAddr::Ip(std::net::SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT)))
+    );
+
+    // Drop mid-pipe (before reply#2 — the quiet-wait phase).
+    drop(session);
+    no_leaked_channels(&topo).await;
+
+    // The connection is healthy: a fresh bind round-trips.
+    let mut fresh = BindSession::bind(
+        topo.consumer.as_ref(),
+        ("fresh.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("fresh bind after the drop");
+    let mut plane = fresh
+        .await_inbound()
+        .await
+        .expect("reply#2")
+        .expect("the plane");
+    plane.write_all(b"again").await.expect("write");
+    plane.flush().await.expect("flush");
+    let mut buf = [0u8; 5];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        plane.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the fresh echo");
+    assert_eq!(&buf, b"again");
+    drop(plane);
+    fresh.close().await;
+    no_leaked_channels(&topo).await;
+}
+
+/// The no-`Clone` posture for the BIND session (the CONNECT session's
+/// compile shape, mirrored): the doctest on the type is the
+/// compile-assert; this probe pins `Send`.
+#[test]
+fn bind_session_type_is_send() {
+    fn assert_send<T: Send>() {}
+    assert_send::<BindSession>();
+    assert_send::<BindReply1<'static>>();
+    assert_send::<BindDataPlane<'static>>();
+}