consumer: AssociateSession — the framed datagram stage + e2e suite

The UDP ASSOCIATE consumer session (POC #2's shape, consumer.md §The
associate session, ADR-003):

- `AssociateSession::associate` — open → no-auth → ASSOCIATE → reply;
  the sentinel BND.ADDR is received (`relay_addr()`, RFC fidelity)
  and never used for addressing (the handle is the channel). The
  ASSOCIATE DST.ADDR is the vanilla-mirror unspecified address.
- `send_datagram` — frames exactly once via the crate's wire
  `frame()` (the framing-once mirror; the u16 joint bound enforced
  at frame time). Takes `impl ToTargetAddr` (ADR-012 posture); a
  `ToTargetAddr for &TargetAddr` impl was added to the forked codec
  so a `TargetAddr` value survives across sends.
- `recv_datagram` — the incremental `DatagramReader` + batched-
  datagram pending queue, state persisting across calls (the POC's
  reader-state finding); `Ok(None)` on channel EOF; fail-loud on a
  stream end mid-frame and on undecodable frames (`WireError` → the
  forked `AddrError`/`UdpHeaderError` RFC classes — the consumer
  mirror of the producer stage's `UndecodableFrame` exit).
- `shutdown` — the half-close; the mux's implicit-EOF sentinel
  reaches the producer's stage as EOF (association lifetime =
  channel lifetime). `close()`/`Drop` own the teardown (sync reap +
  structural EOF, the CONNECT/BIND posture); teardown-at-birth on
  in-stream refusals; no `Clone` (compile-fail asserted).
- Typed error surface: open failures are `SocksSessionError::Open`
  (ADR-049); the missing-egress 0x07 and factory-failure codes
  surface from `associate` as `Rfc`.

E2E suite (`tests/consumer-associate.rs`, 8 tests over the harness):
round-trips (multiple in-order datagrams, one association, sentinel
relay address asserted), the split-datagram persistence test (a 60
KiB datagram reassembling across the call boundary + the pending-
queue drain — both halves of the persistence invariant), the empty-
payload round-trip (F-2, never EOF), shutdown → a fresh associate
succeeds immediately (the POC's test 4), Drop teardown + connection
health, the missing-egress refusal (typed 0x07, no phantom channel),
the egress-factory failure mapping, and the CONNECT+ASSOCIATE
regression guard (interleaved on one registration). Unit tests (4)
pin the reader internals over a duplex (`skip_auth`'s
handshake-free path).

Verification: cargo test 305 (7 suites, all green), clippy
--all-targets -D warnings, fmt, cargo doc --no-deps, and both
wasm32-unknown-unknown checks clean.

src/consumer.rs

@@ -12,32 +12,51 @@
 //! 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`).
+//! then the pass-through data plane. [`AssociateSession`] is the UDP
+//! ASSOCIATE session (POC #2's shape — see its docs).
 //!
 //! 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 CONNECT tears the adopted channel down before the
+//! 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.
+//! 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 target is an RFC reply
-//! code surfaced through [`SocksSessionError::Rfc`].
+//! never yields a phantom channel); a refused ASSOCIATE is an RFC
+//! reply code surfaced through [`SocksSessionError::Rfc`].
 
 use alkcall::channels::client::ChannelClient;
 use alkcall::channels::manager::ChannelManager;
 use alkcall::core::types::BiStream;
+use bytes::Bytes;
+use std::net::{IpAddr, Ipv6Addr, SocketAddr};
 use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf};
 
 use crate::error::{SocksOpenError, SocksSessionError};
@@ -45,7 +64,8 @@ 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};
+use crate::rfc1928::{ReplyError, Socks5Command, SocksError, UdpHeaderError};
+use crate::wire::{frame, DatagramReader, WireError};
 
 /// A completed SOCKS5 CONNECT session on the consumer side: the
 /// adopted channel + the post-reply byte pipe to the target.
@@ -177,6 +197,302 @@ impl Drop for Socks5ChannelSession {
     }
 }
 
+/// 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, 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 relay_addr = inner
+                .request(
+                    Socks5Command::UDPAssociate,
+                    TargetAddr::Ip(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0)),
+                )
+                .await?;
+            Result::<(TargetAddr, Socks5Stream<BiStream>), SocksSessionError>::Ok((
+                relay_addr, inner,
+            ))
+        };
+
+        match negotiated.await {
+            Ok((relay_addr, inner)) => Ok(Self {
+                channel_id,
+                manager: Some(client.manager().clone()),
+                inner: Some(inner),
+                reader: DatagramReader::new(),
+                pending: std::collections::VecDeque::new(),
+                relay_addr: Some(relay_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
+    }
+
+    /// 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.
+    pub fn into_inner(mut self) -> TakenAssociateInner {
+        let inner = self.inner.take();
+        TakenAssociateInner {
+            session: self,
+            inner: inner.expect("into_inner on a spent session"),
+        }
+    }
+
+    /// 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()
+    }
+
+    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 {
@@ -194,7 +510,6 @@ 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
@@ -464,6 +779,28 @@ impl Drop for BindSession {
     }
 }
 
+/// 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
@@ -494,6 +831,7 @@ where
 mod tests {
     use super::*;
     use alkcall::channels::client::ChannelOpenError;
+    use futures::future::FutureExt as _;
 
     fn assert_send<T: Send>() {}
 
@@ -534,4 +872,176 @@ mod tests {
         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());
+    }
+
+    #[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(_))
+        ));
+    }
+
+    /// 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,
+        }
+    }
 }

src/lib.rs

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

src/rfc1928/util/target_addr.rs

@@ -200,6 +200,12 @@ impl ToTargetAddr for TargetAddr {
     }
 }
 
+impl ToTargetAddr for &TargetAddr {
+    fn to_target_addr(&self) -> io::Result<TargetAddr> {
+        Ok((*self).clone())
+    }
+}
+
 #[derive(Debug)]
 pub enum Addr {
     V4([u8; 4]),

tasks/implementation/core-consumer-associate.md

@@ -1,7 +1,7 @@
 ---
 id: implementation/core-consumer-associate
 name: "Core — consumer ASSOCIATE: AssociateSession + e2e tests"
-status: pending
+status: completed
 depends_on: [implementation/core-consumer-connect, implementation/core-producer-associate]
 scope: moderate
 risk: medium
@@ -51,16 +51,16 @@ adapted to the crate's wire codec):
 
 ## Acceptance Criteria
 
-- [ ] `AssociateSession` with `associate`/`send_datagram`/
+- [x] `AssociateSession` with `associate`/`send_datagram`/
       `recv_datagram`/`shutdown` per the POC shape
-- [ ] Reader state persists across `recv_datagram` calls (the
+- [x] Reader state persists across `recv_datagram` calls (the
       split-datagram test)
-- [ ] Framing-once mirror (send frames exactly once; recv decodes
+- [x] Framing-once mirror (send frames exactly once; recv decodes
       incrementally)
-- [ ] Teardown ownership on Drop; no `Clone` (compile-asserted)
-- [ ] E2E tests through the harness (round-trips, shutdown, Drop
+- [x] Teardown ownership on Drop; no `Clone` (compile-asserted)
+- [x] E2E tests through the harness (round-trips, shutdown, Drop
       teardown, regression guard)
-- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean
+- [x] `cargo test` green; clippy/fmt clean; both wasm checks clean
 
 ## References
 
@@ -72,8 +72,87 @@ adapted to the crate's wire codec):
 
 ## Notes
 
-> To be filled by implementation agent
+- **`send_datagram` takes `impl ToTargetAddr`** (the CONNECT
+  session's target-construction posture — ADR-012: IP-string forms
+  parse, domains pass through unresolved), not the POC's
+  `&TargetAddr`: the same entry shape as `connect`. A
+  `ToTargetAddr for &TargetAddr` impl was added to the forked
+  address codec (our owned code, ADR-013) so callers can reuse a
+  `TargetAddr` value across sends.
+- **Teardown ownership matches the CONNECT/BIND sessions** (the
+  crate's posture, strengthened from the POC's "close/join"): the
+  session owns the adopted channel entry — Drop sync-reaps
+  (`teardown_channel`, best-effort; `Drop` cannot await) + the data
+  plane EOFs structurally (the mux's implicit-EOF sentinel);
+  `shutdown()` half-closes (the graceful path, returning the reap
+  result); `close()` is the `shutdown`+reap alias-shaped variant
+  kept for the connect/bind symmetry; teardown-at-birth covers the
+  in-stream refusal path (a refused ASSOCIATE never leaves a
+  phantom channel).
+- **`recv_datagram` returns `Result<Option<(TargetAddr, Bytes)>>`**
+  (the POC's shape): `Ok(None)` = channel EOF (the association's
+  end); a stream end mid-frame is fail-loud (an `UnexpectedEof`
+  I/O error through `SocksSessionError::Rfc` — never a silent
+  partial datagram); an undecodable frame (garbage ATYP) errors
+  loud through the wire codec's `WireError` → RFC mapping
+  (`AddrError` class for address-encoding refusals,
+  `UdpHeaderError` for the garbage shapes — the consumer-side
+  mirror of the producer stage's `UndecodableFrame` exit).
+- **The pending queue stores decoded datagrams** (`(TargetAddr,
+  Bytes)`) — the POC stored pre-decode inner bytes; decoded storage
+  keeps the delivery shape uniform (decode once per frame either
+  way; no wire.rs change needed — `decode_inner` stays private).
+- **Unit tests ride a duplex-backed session** (`skip_auth`'s
+  handshake-free `use_stream` path over a `BiStream::from_joined`
+  duplex — no channels needed) for the reader-internals probes:
+  reader-state persistence (the 2-byte-prefix chunk shape, the POC
+  finding's trigger), mid-frame EOF fail-loud, garbage-ATYP fail-
+  loud, and the wire-error mapping table.
+- **The e2e split-datagram test uses a 60 KiB payload** (60007
+  inner bytes — legal under the u16 joint bound, well over the 16
+  KB read buffer): first → big → third, in order, with the big
+  datagram reassembling across the call boundary and the third
+  drained through the pending queue — both halves of the
+  persistence invariant pinned end to end.
+- **No harness changes**: the consumer-side topology's
+  `wire_consumer_side_with` + the producer-associate task's
+  `echo_egress`/`refusing_egress` closures cover every test.
+- The lib.rs re-exports gained `AssociateSession` +
+  `TakenAssociateInner`; the forked `target_addr.rs` gained the
+  `&TargetAddr` `ToTargetAddr` impl (ergonomics; owned code,
+  ADR-013).
 
 ## Summary
 
-> To be filled on completion
+Implemented the UDP ASSOCIATE consumer session in `src/consumer.rs`
+(POC #2's shape, adapted to the crate's teardown-owning session
+posture): `AssociateSession::associate` (open → no-auth → ASSOCIATE
+→ reply, capturing the sentinel BND.ADDR as `relay_addr()` —
+received, never used for addressing; the ASSOCIATE DST.ADDR is the
+vanilla-mirror unspecified address), `send_datagram` (frames once
+via the crate's wire `frame()` — the framing-once mirror, the joint
+bound enforced at frame time), `recv_datagram` (the incremental
+`DatagramReader` + batched-datagram pending queue, **state that
+persists across calls** — the POC's reader-state finding, with the
+fail-loud shapes: mid-frame EOF and undecodable frames), `shutdown`
+(the half-close; the mux's implicit-EOF sentinel reaches the
+producer's stage as EOF — association lifetime = channel lifetime),
+plus `split`/`into_inner` accessors and the CONNECT sessions'
+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
+missing-egress 0x07 and factory-failure codes surface from
+`associate` as `Rfc`. E2E suite (`tests/consumer-associate.rs`, 8
+tests over the harness): the round-trip (sentinel relay address →
+multiple in-order datagrams, one association), the split-datagram
+persistence test (60 KiB datagram across the call boundary +
+pending-queue drain), empty-payload round-trip (F-2, never EOF),
+shutdown → fresh associate succeeds immediately (the POC's test 4),
+Drop teardown + connection health, the missing-egress refusal
+(typed 0x07, no phantom channel), the egress-factory failure
+mapping, and the CONNECT+ASSOCIATE regression guard on one
+registration (interleaved, both serving). Verification: `cargo
+test` 305 (incl. the 3 doctests — the compile-fail assertions ride
+among them), clippy `-D warnings` (all targets), fmt, `cargo doc
+--no-deps`, and both wasm checks clean.

tests/consumer-associate.rs

@@ -0,0 +1,479 @@
+//! The consumer UDP ASSOCIATE integration tests — `AssociateSession`
+//! end to end through the harness's consumer-side topology (the
+//! producer's real command-dispatch path over channels; the framed
+//! datagram stage, ADR-003, driven by the wrapper-aware client, POC
+//! #2's shape).
+//!
+//! Covered: associate → sentinel reply (received, never used for
+//! addressing) → send/recv round-trips (multiple in-order datagrams,
+//! one association); the split-datagram persistence test (reader
+//! state persists across `recv_datagram` calls — the POC's
+//! 2-byte-prefix chunk finding); empty-payload datagram round-trip
+//! (F-2, never EOF); shutdown → the producer's stage ends and a
+//! fresh associate succeeds immediately; Drop teardown (drop
+//! mid-stream → the producer's stage sees EOF, the connection stays
+//! healthy); the ASSOCIATE refusal surfaced as a typed error with no
+//! phantom channel; CONNECT + ASSOCIATE on the same registration
+//! (the regression guard extended).
+
+mod harness;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use alksocks::consumer::{AssociateSession, Socks5ChannelSession};
+use alksocks::egress::EgressError;
+use alksocks::error::SocksSessionError;
+use alksocks::rfc1928::consts;
+use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
+use alksocks::SocksError as SocksErrorKind;
+
+use harness::{echo_dial, echo_egress, refusing_egress, wire_consumer_side_with};
+
+/// The no-leaked-channels probe (the connect/bind suites' shape,
+/// re-pinned for the ASSOCIATE session): 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");
+}
+
+/// The happy path: `associate` → the reply's sentinel `BND.ADDR` is
+/// received (RFC fidelity) and never used for addressing → send/recv
+/// round-trips through the echo egress — multiple datagrams, in
+/// order, on one association; the establisher witnessed the opener
+/// identity (CF-006).
+#[tokio::test]
+async fn associate_round_trips_multiple_datagrams_one_association() {
+    let topo = wire_consumer_side_with(
+        echo_dial(),
+        Some(echo_egress()),
+        None,
+        Some(harness::consumer_identity()),
+    )
+    .await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+
+    // The reply's BND.ADDR is the producer's sentinel (127.0.0.1:0 —
+    // ADR-003): received for RFC fidelity, never used for addressing.
+    let relay = assoc
+        .relay_addr()
+        .expect("the reply carried BND.ADDR")
+        .into_string_and_port();
+    assert_eq!(relay.0, "127.0.0.1", "the in-band sentinel BND.ADDR");
+    assert_eq!(relay.1, 0);
+
+    let target = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
+    for i in 0..5u8 {
+        let payload = vec![i; 10];
+        assoc
+            .send_datagram(&target, &payload)
+            .await
+            .expect("send datagram");
+    }
+    for i in 0..5u8 {
+        let (src, payload) =
+            tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+                .await
+                .expect("recv within timeout")
+                .expect("a datagram (not eof)")
+                .expect("a datagram");
+        assert_eq!(src, target, "the echo reports the destination");
+        assert_eq!(payload, bytes::Bytes::from(vec![i; 10]), "in-order echo");
+    }
+
+    // The association is alive: another datagram round-trips.
+    assoc.send_datagram(&target, b"alive").await.expect("send");
+    let (_, payload) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&payload[..], b"alive");
+
+    assert!(
+        topo.consumer.manager().has_channel(assoc.channel_id()),
+        "the adopted channel entry exists while associated"
+    );
+
+    let reaped = assoc.shutdown().await.expect("half-close");
+    assert!(reaped, "the adopted channel entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// The split-datagram persistence test (POC #2's finding — the
+/// reader state must persist across `recv_datagram` calls): a large
+/// datagram's frame (under the u16 joint bound but well over the 16
+/// KB read buffer) spans multiple channel reads AND multiple
+/// `recv_datagram` calls — the first call returns the earlier small
+/// datagram while the large one is mid-flight, the next call
+/// reassembles it across its internal reads — and a trailing small
+/// datagram batched in the same read batch drains through the
+/// pending queue.
+#[tokio::test]
+async fn recv_datagram_state_persists_across_calls() {
+    let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+    let target = TargetAddr::Ip("10.0.0.2:53".parse().unwrap());
+
+    // The order tells the story: "first" completes early; the 60 KiB
+    // datagram (60007 inner bytes — legal, but over the 16 KB read
+    // buffer) is mid-flight when the first call returns; "third"
+    // batches after it.
+    assoc
+        .send_datagram(&target, b"first")
+        .await
+        .expect("send the first datagram");
+    let big = vec![7u8; 60 * 1024];
+    assoc
+        .send_datagram(&target, &big)
+        .await
+        .expect("send the big datagram");
+    assoc
+        .send_datagram(&target, b"third")
+        .await
+        .expect("send the third datagram");
+
+    let (src, first) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(src, target);
+    assert_eq!(&first[..], b"first", "the first datagram, in order");
+
+    // The large datagram reassembles across the call boundary (its
+    // mid-frame state persisted into this call) — the POC finding's
+    // shape.
+    let (src, second) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(src, target);
+    assert_eq!(&second[..], &big[..], "the split datagram reassembled");
+
+    // The trailing datagram rode the same read batches: the pending
+    // queue carries it to this call.
+    let (_, third) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&third[..], b"third", "the batched datagram drained");
+
+    let reaped = assoc.shutdown().await.expect("half-close");
+    assert!(reaped);
+    no_leaked_channels(&topo).await;
+}
+
+/// An empty-payload datagram is a legal datagram — never EOF (the
+/// F-2 invariant, end to end through the echo egress), and the
+/// association survives it (pre → empty → post, in order).
+#[tokio::test]
+async fn empty_payload_datagram_round_trips_and_survives() {
+    let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+    let target = TargetAddr::Ip("10.0.0.3:53".parse().unwrap());
+
+    assoc
+        .send_datagram(&target, b"pre")
+        .await
+        .expect("send pre");
+    assoc
+        .send_datagram(&target, b"")
+        .await
+        .expect("an empty payload is a legal datagram");
+    assoc
+        .send_datagram(&target, b"post")
+        .await
+        .expect("send post");
+
+    let (_, pre) = tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("a datagram (not eof)")
+        .expect("a datagram");
+    assert_eq!(&pre[..], b"pre");
+
+    let (src, empty) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(src, target);
+    assert!(
+        empty.is_empty(),
+        "the empty-payload datagram round-trips (never EOF)"
+    );
+
+    let (_, post) = tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("a datagram (not eof)")
+        .expect("a datagram");
+    assert_eq!(&post[..], b"post");
+
+    let reaped = assoc.shutdown().await.expect("half-close");
+    assert!(reaped);
+    no_leaked_channels(&topo).await;
+}
+
+/// 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) — and a fresh associate
+/// succeeds immediately after (the POC's test 4, from the consumer
+/// side; the same registration is unpoisoned).
+#[tokio::test]
+async fn shutdown_ends_the_association_and_a_fresh_associate_succeeds() {
+    let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+    let channel_id = assoc.channel_id();
+    let target = TargetAddr::Ip("10.0.0.4:53".parse().unwrap());
+    assoc
+        .send_datagram(&target, b"before")
+        .await
+        .expect("send before shutdown");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"before");
+
+    // Half-close: the producer's stage exits on the EOF sentinel.
+    assoc.shutdown().await.expect("shutdown");
+    assert!(
+        !topo.consumer.manager().has_channel(channel_id),
+        "the adopted channel entry was reaped"
+    );
+
+    // The connection is healthy: a fresh associate round-trips.
+    let mut fresh = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("fresh associate after the shutdown");
+    fresh
+        .send_datagram(&target, b"fresh")
+        .await
+        .expect("send after fresh associate");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), fresh.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"fresh");
+    let reaped = fresh.shutdown().await.expect("shutdown");
+    assert!(reaped);
+    no_leaked_channels(&topo).await;
+}
+
+/// Drop teardown: dropping a live association mid-stream reaps the
+/// adopted channel entry (no leak), the producer's stage sees the
+/// structural EOF (the mux's implicit-EOF sentinel on the dropped
+/// write half), and the connection stays healthy — a fresh associate
+/// succeeds immediately after.
+#[tokio::test]
+async fn drop_teardown_reaps_and_connection_stays_healthy() {
+    let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+    let channel_id = assoc.channel_id();
+    assoc
+        .send_datagram(TargetAddr::Ip("10.0.0.5:53".parse().unwrap()), b"mid")
+        .await
+        .expect("send before the drop");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"mid");
+
+    // Drop mid-stream (the teardown-ownership posture).
+    drop(assoc);
+    assert!(
+        !topo.consumer.manager().has_channel(channel_id),
+        "the adopted channel entry was reaped on Drop"
+    );
+    no_leaked_channels(&topo).await;
+
+    // The connection is healthy: a fresh associate round-trips.
+    let mut fresh = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("fresh associate after the drop");
+    let target = TargetAddr::Ip("10.0.0.5:53".parse().unwrap());
+    fresh
+        .send_datagram(&target, b"again")
+        .await
+        .expect("send after the drop");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), fresh.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"again");
+    fresh.shutdown().await.expect("shutdown");
+    no_leaked_channels(&topo).await;
+}
+
+/// A missing egress seam: the ASSOCIATE refusal surfaces the RFC
+/// reply code through the session type (`CommandNotSupported` 0x07 —
+/// ADR-008's missing-callback class), and no phantom channel remains
+/// on either side (teardown-at-birth).
+#[tokio::test]
+async fn missing_egress_refusal_surfaces_rfc_reply_code_no_phantom_channel() {
+    let topo = wire_consumer_side_with(echo_dial(), None, None, None).await;
+
+    let err = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect_err("no egress wired — the ASSOCIATE must refuse");
+
+    match &err {
+        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => {
+            assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED);
+        }
+        other => panic!("expected an Rfc reply-code refusal, got {other:?}"),
+    }
+
+    no_leaked_channels(&topo).await;
+
+    // The connection is healthy: wiring the egress serves ASSOCIATE.
+    let topo2 = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+    let mut assoc = AssociateSession::associate(topo2.consumer.as_ref())
+        .await
+        .expect("fresh associate after the refusal");
+    assoc
+        .send_datagram(TargetAddr::Ip("10.0.0.6:53".parse().unwrap()), b"fresh")
+        .await
+        .expect("send");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"fresh");
+    assoc.shutdown().await.expect("shutdown");
+    no_leaked_channels(&topo2).await;
+}
+
+/// An egress-factory failure: the mapped RFC reply code surfaces from
+/// `associate` (reply#1's code — the factory refused before the
+/// reply), and no phantom channel remains.
+#[tokio::test]
+async fn egress_factory_refusal_surfaces_the_mapped_reply_code() {
+    let topo = wire_consumer_side_with(
+        echo_dial(),
+        Some(refusing_egress(EgressError::Unreachable(
+            "no egress".into(),
+        ))),
+        None,
+        None,
+    )
+    .await;
+
+    let err = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect_err("the egress factory refused");
+    match &err {
+        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => {
+            assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_NETWORK_UNREACHABLE);
+        }
+        other => panic!("expected an Rfc reply-code refusal, got {other:?}"),
+    }
+    no_leaked_channels(&topo).await;
+}
+
+/// CONNECT + ASSOCIATE on the same registration (the regression
+/// guard extended): one ALPN, both commands, interleaved — an
+/// association and a CONNECT session live concurrently on the same
+/// registration, both serving.
+#[tokio::test]
+async fn connect_and_associate_served_on_the_same_registration() {
+    let topo = wire_consumer_side_with(echo_dial(), Some(echo_egress()), None, None).await;
+
+    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
+        .await
+        .expect("the association associates");
+    let mut connect = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("echo.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the CONNECT session connects");
+
+    // The association round-trips while the CONNECT pipe is open.
+    let target = TargetAddr::Ip("10.0.0.7:53".parse().unwrap());
+    assoc
+        .send_datagram(&target, b"udp")
+        .await
+        .expect("send datagram");
+    let (_, echoed) =
+        tokio::time::timeout(std::time::Duration::from_secs(10), assoc.recv_datagram())
+            .await
+            .expect("recv timed out")
+            .expect("a datagram (not eof)")
+            .expect("a datagram");
+    assert_eq!(&echoed[..], b"udp");
+
+    // The CONNECT pipe round-trips while the association is open.
+    connect
+        .split()
+        .expect("pipe")
+        .1
+        .write_all(b"tcp")
+        .await
+        .expect("write");
+    let (mut read, _) = connect.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");
+
+    assoc.shutdown().await.expect("shutdown");
+    connect.close().await;
+    no_leaked_channels(&topo).await;
+}