tests: the review-002 coverage residue — stage exits, splits, spent arms (N-8)

Eighteen new tests, zero production-code changes; every line the
review attributed in §N-8 is now covered (verified by a
`cargo llvm-cov --show-missing-lines` before/after diff).

- producer: the datagram stage's failure exits pinned as direct unit
  tests on `run_datagram_stage` (the serve layer discards StageExit,
  so the unit shape is the only place the exact matrix is assertable)
  — `ReplyFrameRejected` (a joint-bound-violating reply), `EgressEnded`
  via `next_reply`'s Err, `Io` via a failing write and a failing read;
  the `send`-failure arm got the serve-layer drop-witness treatment
  (full ASSOCIATE conversation, EOF after the refusal, the drop flag
  flips — teardown-at-birth on the failure path)
- producer: the quiet wait's `Err(_)` arm (a read *error* ends the
  wait `Ok(())`, the accept future drops with its owned state — the
  client-gone contract's error-shaped twin; driven directly on
  `bind_quiet_wait_and_serve`) and the BIND reply#2 Domain-peer
  fallback (`TargetAddr::Domain(_, port)` → `0.0.0.0:port`, the data
  plane live end to end)
- producer: the establisher's `None`-witness arm (the registration's
  optional witness seam) added to the existing unit test
- consumer: the two front-door-bridge `split`s tested on live
  `Socks5Stream<BiStream>` streams via `build_skip_auth_bi_stream`
  (two duplexes joined the `open_and_negotiate` shape) — raw framed
  bytes through both halves; the BIND split leaves the take-guard
  untouched (`await_inbound` still reads the reply#2 phase)
- consumer: the single-arm residue — spent `send_datagram`'s `Io` arm
  (+ the spent-`recv` `Ok(None)` shape asserted alongside), the
  mid-batch later-frame `Err` dropping the pending `Ok`s (the
  fail-loud-drops-pending semantics), ASSOCIATE `into_inner`'s `Ok`
  pairing, `BindDataPlane`'s `poll_flush`/`poll_shutdown`
  passthroughs (the BIND door's half-close), `read_reply2`'s
  version pin (a foreign producer's non-5 reply#2), the
  ASSOCIATE/BIND session `Debug` bodies, the spent-`close` no-op
- egress/error: the three seam-handle `Debug` renders
  (TargetHandle / BindTargetHandle / BindAccept) and
  `SocksEstablishmentReason`'s `Display` (each variant renders as
  its canonical wire code)

Coverage: 96.31% regions / 97.22% lines; no new accepted-uncovered
entries (the residual uncovered lines are the pre-existing
out-of-scope set + producer.rs 188-189's untestable `accept_bi`
substrate-failure log arm). One test-only select-shape discovery
recorded in the task Notes: `tokio::select!`'s random branch order
can drop a not-yet-polled accept future before its owned state
exists — the quiet-wait test parks once before erroring so both
branches are guaranteed started.

Verification: `cargo test` + `--all-features` green (375 tests),
clippy `-D warnings` clean (native + wasm), fmt clean, `cargo doc`
clean, both wasm checks clean.

src/consumer.rs

@@ -1158,6 +1158,343 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn send_datagram_on_a_spent_stream_is_the_io_arm() {
+        // The spent-stream `Io` arm (review 002 N-8): a session whose
+        // stream is gone errors loud through the typed I/O surface —
+        // the mirror of the covered spent-`recv` arm.
+        let mut session = AssociateSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: None,
+        };
+        let err = session
+            .send_datagram(TargetAddr::Ip("10.0.0.1:53".parse().unwrap()), b"payload")
+            .await
+            .expect_err("a spent stream must not send");
+        assert!(
+            matches!(err, SocksSessionError::Io(_)),
+            "the spent-stream send is the Io arm: {err:?}"
+        );
+
+        // The spent-session read shape rides the same posture: the
+        // `recv_datagram` spent arm is `Ok(None)` (the EOF-shape, the
+        // family's read-side posture).
+        assert!(
+            session.recv_datagram().await.expect("spent recv").is_none(),
+            "a spent stream reads the spent shape, never an error"
+        );
+
+        // The spent-session teardown surfaces: `shutdown` on a spent
+        // session is a no-op reap (no manager — nothing reaped).
+        let shutdown = session.shutdown().await.expect("spent shutdown");
+        assert!(!shutdown, "no manager — nothing reaped");
+    }
+
+    #[tokio::test]
+    async fn spent_associate_close_reaps_a_no_op() {
+        // The `close`-on-spent arm: the graceful teardown path is a
+        // no-op reap on a spent session (the session family's Drop
+        // posture, on the async path).
+        let session = AssociateSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: None,
+        };
+        assert!(!session.close().await, "no manager — nothing reaped");
+    }
+
+    #[tokio::test]
+    async fn recv_datagram_mid_batch_later_frame_err_drops_the_pending_oks() {
+        use tokio::io::AsyncWriteExt as _;
+
+        // The mid-batch later-frame `Err` arm (review 002 N-8): the
+        // first frame of a batch decodes `Ok`, a later one is garbage
+        // — the batch is discarded with the session (fail-loud, never
+        // a partial batch delivered while the session stays open).
+        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
+        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
+        let mut wire = crate::wire::frame(&addr, b"first").unwrap().to_vec();
+        wire.extend_from_slice(&[0x00, 0x04, 0x7f, 1, 2, 3]);
+        peer_writer.write_all(&wire).await.unwrap();
+        drop(peer_writer);
+
+        let (write_peer, session_writer) = tokio::io::duplex(64);
+        let _ = write_peer;
+
+        let mut session = AssociateSession {
+            channel_id: 1,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: None,
+        };
+        let err = session
+            .recv_datagram()
+            .await
+            .expect_err("the later frame's garbage fails loud");
+        assert!(
+            matches!(err, SocksSessionError::Rfc(SocksError::UdpHeaderError(_))),
+            "the decode garbage rides the UDP-header class: {err:?}"
+        );
+
+        // The pending `Ok` was dropped with the session — the queue
+        // stays empty and the stream is spent for further reads (the
+        // error ended the caller's loop, the fail-loud posture).
+        assert!(session.pending.is_empty());
+    }
+
+    #[tokio::test]
+    async fn associate_into_inner_pairs_the_stream_with_the_teardown_owner() {
+        // The `Ok` arm of `AssociateSession::into_inner` (review 002
+        // N-8): the taken stream pairs with the session that stays the
+        // teardown owner (the `into_inner_pairs_the_pipe_with_the_
+        // teardown_owner` shape's ASSOCIATE mirror) — and a second
+        // take is the spent shape.
+        let (_write_peer, session_writer) = tokio::io::duplex(64);
+        let (read_peer, session_reader) = tokio::io::duplex(64);
+        let _ = read_peer;
+
+        let session = AssociateSession {
+            channel_id: 1,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())),
+        };
+
+        let taken = session.into_inner().expect("the stream is live");
+        // The pair carries both halves: the teardown owner (the relay
+        // address still observable) and the owned stream.
+        assert!(taken.session.relay_addr().is_some(), "the owner");
+        assert!(taken.session.channel_id() == 1);
+        drop(taken);
+    }
+
+    #[tokio::test]
+    async fn associate_debug_renders_channel_liveness_and_pending() {
+        // The Debug impl body (review 002 N-8, egress.rs class): the
+        // associate session renders its channel id, liveness, the
+        // relay address, and the pending count.
+        let mut session = AssociateSession {
+            channel_id: 7,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(
+                tokio::io::duplex(16).0,
+                tokio::io::duplex(16).0,
+            )),
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: Some(TargetAddr::Ip("127.0.0.1:0".parse().unwrap())),
+        };
+        session.pending.push_back((
+            TargetAddr::Ip("10.0.0.1:53".parse().unwrap()),
+            Bytes::from_static(b"x"),
+        ));
+        let rendered = format!("{session:?}");
+        assert!(rendered.contains("channel_id: 7"), "{rendered}");
+        assert!(rendered.contains("associated: true"), "{rendered}");
+        assert!(rendered.contains("relay_addr"), "{rendered}");
+        assert!(rendered.contains("pending_datagrams: 1"), "{rendered}");
+        drop(session);
+
+        let bind = BindSession {
+            channel_id: 8,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(
+                tokio::io::duplex(16).0,
+                tokio::io::duplex(16).0,
+            )),
+            bind_addr: Some(TargetAddr::Domain("bind.test".into(), 2121)),
+        };
+        let rendered = format!("{bind:?}");
+        assert!(rendered.contains("channel_id: 8"), "{rendered}");
+        assert!(rendered.contains("awaiting_inbound: true"), "{rendered}");
+        assert!(rendered.contains("bind_addr"), "{rendered}");
+    }
+
+    #[tokio::test]
+    async fn associate_split_yields_live_halves_and_survives_the_round_trip() {
+        use crate::wire::frame as wire_frame;
+        use tokio::io::AsyncWriteExt as _;
+
+        // The front-door-bridge surface (consumer.md) on a *live*
+        // stream: the halves borrow the session and the raw framed
+        // bytes pass through both directions (0 B framing — the door's
+        // RFC header codec would own the frames at the boundary).
+        let addr = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());
+        let framed = wire_frame(&addr, b"split").unwrap().to_vec();
+
+        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
+        peer_writer.write_all(&framed).await.unwrap();
+        drop(peer_writer);
+        let (mut peer_reader, session_writer) = tokio::io::duplex(1024);
+
+        let mut session = AssociateSession {
+            channel_id: 1,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
+            reader: DatagramReader::new(),
+            pending: Default::default(),
+            relay_addr: None,
+        };
+        let (mut read, mut write) = session.split().expect("live halves");
+
+        // Read the raw frame through the borrowed read half.
+        let mut header = [0u8; 2];
+        read.read_exact(&mut header)
+            .await
+            .expect("the frame header");
+        let len = u16::from_be_bytes(header) as usize;
+        let mut rest = vec![0u8; len];
+        read.read_exact(&mut rest).await.expect("the frame body");
+        assert_eq!(len, 7 + 5, "one v4-address datagram frame");
+        drop(read);
+
+        // The write half passes through (the bridge seam's shape) —
+        // the peer receives the exact bytes.
+        write.write_all(&framed).await.unwrap();
+        write.flush().await.unwrap();
+        drop(write);
+
+        let mut got = vec![0u8; framed.len()];
+        peer_reader
+            .read_exact(&mut got)
+            .await
+            .expect("the peer read");
+        assert_eq!(got, framed, "the write half rode the raw pipe");
+    }
+
+    /// An inner `Socks5Stream` over a true `BiStream` (the session/
+    /// plane field type) for the BIND-side surfaces: reads come from
+    /// `session_end` (the peer's writes), writes go out via the duplex
+    /// whose other end the test holds. Two joined halves, the same
+    /// shape `open_and_negotiate` assembles.
+    fn build_skip_auth_bi_stream(
+        session_end: tokio::io::DuplexStream,
+        writer_end: tokio::io::DuplexStream,
+    ) -> Socks5Stream<BiStream> {
+        let mut config = Config::default();
+        config.set_skip_auth(true);
+        Socks5Stream::use_stream(BiStream::from_joined(session_end, writer_end), None, config)
+            .now_or_never()
+            .expect("skip_auth's use_stream is instantly ready")
+            .expect("skip_auth's use_stream succeeds")
+    }
+
+    #[tokio::test]
+    async fn bind_split_yields_live_halves_pre_reply2() {
+        use tokio::io::AsyncWriteExt as _;
+
+        // The BIND front-door-bridge surface (consumer.md): the split
+        // halves in the reply#1 phase — pre-reply#2 bytes ride the
+        // same channel (the doc's claim, exercised on a live stream).
+        let (mut peer_writer, session_reader) = tokio::io::duplex(1024);
+        peer_writer.write_all(b"raw").await.unwrap();
+        drop(peer_writer);
+
+        let (mut peer_reader, session_writer) = tokio::io::duplex(1024);
+        let mut session = BindSession {
+            channel_id: 1,
+            manager: None,
+            inner: Some(build_skip_auth_bi_stream(session_reader, session_writer)),
+            bind_addr: Some(TargetAddr::Ip("127.0.0.1:2121".parse().unwrap())),
+        };
+        {
+            let (mut read, mut write) = session.split().expect("live pre-reply2 halves");
+            write.write_all(b"ack").await.unwrap();
+            write.flush().await.expect("flush");
+            let mut buf = [0u8; 3];
+            read.read_exact(&mut buf).await.expect("the raw bytes");
+            assert_eq!(&buf, b"raw");
+        }
+
+        // The write half reached the peer (the bridge seam's shape).
+        let mut buf = [0u8; 3];
+        peer_reader.read_exact(&mut buf).await.expect("the ack");
+        assert_eq!(&buf, b"ack");
+        drop(peer_reader);
+
+        // The take-guard is untouched: the reader still reads the
+        // reply#2 phase (the split halves were a view, not a
+        // consumer) — the dropped peer EOFs it loud, never the spent
+        // shape.
+        let outcome = session.await_inbound().await;
+        assert!(
+            !matches!(outcome, Ok(None)),
+            "the reply#2 phase was not consumed by the split"
+        );
+        session.close().await;
+    }
+
+    #[tokio::test]
+    async fn bind_data_plane_poll_flush_and_poll_shutdown_pass_through() {
+        use tokio::io::AsyncWriteExt as _;
+
+        // The BIND door's half-close rides the plane's poll_shutdown
+        // passthrough (and poll_flush rides the write path): write →
+        // flush → shutdown — the bytes reach the peer and the EOF
+        // sentinel follows.
+        let (_session_writer_unused, session_reader) = tokio::io::duplex(64);
+        let (mut peer_reader, session_writer) = tokio::io::duplex(64);
+        let mut plane = BindDataPlane {
+            inner: build_skip_auth_bi_stream(session_reader, session_writer),
+            peer: TargetAddr::Ip("10.0.0.7:44444".parse().unwrap()),
+        };
+        plane.write_all(b"half").await.expect("write");
+        plane.flush().await.expect("the poll_flush passthrough");
+        plane
+            .shutdown()
+            .await
+            .expect("the poll_shutdown passthrough");
+        drop(plane);
+
+        let mut seen = Vec::new();
+        let mut buf = [0u8; 4];
+        loop {
+            match peer_reader.read(&mut buf).await {
+                Ok(0) | Err(_) => break,
+                Ok(n) => seen.extend_from_slice(&buf[..n]),
+            }
+        }
+        assert_eq!(&seen, b"half", "the write rode through, then EOF");
+    }
+
+    #[tokio::test]
+    async fn read_reply2_rejects_a_foreign_producers_version() {
+        use tokio::io::AsyncWriteExt as _;
+
+        // A foreign producer's reply#2 with a non-5 version: the
+        // version pin is loud (never an aliased peer).
+        let (mut peer_writer, session_reader) = tokio::io::duplex(64);
+        peer_writer
+            .write_all(&[0x04, 0x00, 0x00, 0x01, 10, 0, 0, 1, 0, 80])
+            .await
+            .unwrap();
+        drop(peer_writer);
+
+        let (_write_peer_unused, session_writer) = tokio::io::duplex(64);
+        let mut inner = build_skip_auth_bi_stream(session_reader, session_writer);
+        let err = read_reply2(&mut inner)
+            .await
+            .expect_err("a foreign version is refused");
+        assert!(
+            matches!(
+                err,
+                SocksSessionError::Rfc(SocksError::UnsupportedSocksVersion(0x04))
+            ),
+            "the version pin is loud: {err:?}"
+        );
+    }
+
     /// A session over a raw duplex (the reader internals' test
     /// harness — the handshake-free path: `skip_auth` bypasses the
     /// greeting, no reply is expected, the datagram stage rides

src/egress.rs

@@ -592,4 +592,41 @@ mod tests {
         assert_send_sync::<BindAcceptError>();
         assert_send_sync::<BindAcceptFactoryError>();
     }
+
+    #[test]
+    fn debug_impls_render_the_handle_surfaces() {
+        // The three Debug impl bodies (review 002 N-8): the dialed
+        // handle renders non-exhaustive; the accept handle carries the
+        // peer; the accept source carries the bind address.
+        let dialed = TargetHandle {
+            read: Box::new(tokio::io::empty()),
+            write: Box::new(tokio::io::sink()),
+        };
+        let rendered = format!("{dialed:?}");
+        assert!(rendered.contains("TargetHandle"), "{rendered}");
+        assert!(rendered.contains(".."), "{rendered}");
+
+        let accepted = BindTargetHandle {
+            peer: TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444))),
+            read: Box::new(tokio::io::empty()),
+            write: Box::new(tokio::io::sink()),
+        };
+        let rendered = format!("{accepted:?}");
+        assert!(rendered.contains("BindTargetHandle"), "{rendered}");
+        assert!(
+            rendered.contains("10.0.0.7:44444"),
+            "the peer is named: {rendered}"
+        );
+
+        let accept = BindAccept::new(
+            SocketAddr::from(([127, 0, 0, 1], 2121)),
+            std::future::pending::<Result<BindTargetHandle, BindAcceptError>>(),
+        );
+        let rendered = format!("{accept:?}");
+        assert!(rendered.contains("BindAccept"), "{rendered}");
+        assert!(
+            rendered.contains("127.0.0.1:2121"),
+            "the bind address is named: {rendered}"
+        );
+    }
 }

src/error.rs

@@ -227,4 +227,25 @@ mod tests {
         ));
         assert!(matches!(io, SocksSessionError::Io(_)));
     }
+
+    #[test]
+    fn establishment_reason_display_renders_the_wire_code() {
+        // The Display impl body (review 002 N-8): the typed reason
+        // renders as its canonical wire code (the `as_str` inverse).
+        for (typed, wire) in [
+            (SocksEstablishmentReason::DialFailed, "dial_failed"),
+            (
+                SocksEstablishmentReason::UnknownResource,
+                "unknown_resource",
+            ),
+            (
+                SocksEstablishmentReason::ResourceShortage,
+                "resource_shortage",
+            ),
+            (SocksEstablishmentReason::HandlerError, "handler_error"),
+            (SocksEstablishmentReason::Timeout, "timeout"),
+        ] {
+            assert_eq!(typed.to_string(), wire);
+        }
+    }
 }

src/producer.rs

@@ -753,6 +753,8 @@ mod tests {
     use super::*;
     use crate::egress::{BindAcceptError, BindAcceptFactoryError, SocksDialError, TargetHandle};
     use bytes::Bytes;
+    use std::pin::Pin;
+    use std::task::{Context, Poll};
     use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
 
     /// The test dial closure (the harness's echo shape, inline): one
@@ -1626,6 +1628,296 @@ mod tests {
         );
     }
 
+    // -- the datagram stage's failure exits (review 002 N-8) -----------
+    // The stage's teardown surface, driven directly: each failure exit
+    // returns its exact `StageExit` (the serve layer discards it, so
+    // the unit shape is the only place the matrix is pinnable).
+
+    /// A stream half that fails its reads or writes on demand (the
+    /// stage's `Io`-exit triggers).
+    struct FailStream {
+        fail_read: bool,
+        fail_write: bool,
+    }
+
+    impl tokio::io::AsyncRead for FailStream {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &mut tokio::io::ReadBuf<'_>,
+        ) -> Poll<std::io::Result<()>> {
+            if self.fail_read {
+                Poll::Ready(Err(std::io::Error::other("stage read failed")))
+            } else {
+                Poll::Pending
+            }
+        }
+    }
+
+    impl tokio::io::AsyncWrite for FailStream {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            buf: &[u8],
+        ) -> Poll<std::io::Result<usize>> {
+            if self.fail_write {
+                Poll::Ready(Err(std::io::Error::other("stage write failed")))
+            } else {
+                Poll::Ready(Ok(buf.len()))
+            }
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    /// An egress yielding one pre-set reply then parking forever (the
+    /// reply branch wins the select once; the stage never advances
+    /// past it).
+    struct OneReplyEgress {
+        reply: Option<Bytes>,
+    }
+
+    impl crate::egress::Egress for OneReplyEgress {
+        fn send(
+            &mut self,
+            _addr: TargetAddr,
+            _payload: Bytes,
+        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
+            Box::pin(async move { Err(EgressError::General("send refused".into())) })
+        }
+
+        fn next_reply(
+            &mut self,
+        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+            let reply = self.reply.take();
+            Box::pin(async move {
+                match reply {
+                    Some(inner) => Ok(Some(inner)),
+                    None => Ok(std::future::pending::<Option<Bytes>>().await),
+                }
+            })
+        }
+    }
+
+    /// An egress whose relay already ended (`next_reply` errs) — the
+    /// `EgressEnded`-via-`next_reply` trigger.
+    struct RelayEndedEgress;
+
+    impl crate::egress::Egress for RelayEndedEgress {
+        fn send(
+            &mut self,
+            _addr: TargetAddr,
+            _payload: Bytes,
+        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
+            Box::pin(async move { Err(EgressError::General("send refused".into())) })
+        }
+
+        fn next_reply(
+            &mut self,
+        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+            Box::pin(async move { Err(EgressError::General("relay ended".into())) })
+        }
+    }
+
+    /// An egress that parks forever on `next_reply` (the read branch's
+    /// exits are then deterministic — the parked-accept shape's egress
+    /// mirror).
+    struct ParkedEgress;
+
+    impl crate::egress::Egress for ParkedEgress {
+        fn send(
+            &mut self,
+            _addr: TargetAddr,
+            _payload: Bytes,
+        ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
+            Box::pin(async move { Err(EgressError::General("send refused".into())) })
+        }
+
+        fn next_reply(
+            &mut self,
+        ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+            Box::pin(async move { Ok(std::future::pending::<Option<Bytes>>().await) })
+        }
+    }
+
+    /// The `ReplyFrameRejected` exit: an egress reply whose inner
+    /// bytes violate the u16 joint bound — the wire codec refuses it
+    /// at `frame_inner`, the stage tears the association down (never a
+    /// wire overflow).
+    #[tokio::test]
+    async fn datagram_stage_reply_frame_rejected_exit() {
+        let mut egress = Box::new(OneReplyEgress {
+            reply: Some(Bytes::from(vec![0u8; crate::wire::MAX_INNER_LEN + 1])),
+        }) as Box<dyn Egress + Send>;
+        let exit = run_datagram_stage(
+            FailStream {
+                fail_read: false,
+                fail_write: true,
+            },
+            egress.as_mut(),
+        )
+        .await;
+        assert_eq!(exit, StageExit::ReplyFrameRejected);
+    }
+
+    /// The `EgressEnded` exit via `next_reply`'s `Err` — the egress
+    /// relay ended; the association ends with it (the lossy-datagram
+    /// posture).
+    #[tokio::test]
+    async fn datagram_stage_egress_relay_ended_exit_via_next_reply() {
+        let mut egress = Box::new(RelayEndedEgress) as Box<dyn Egress + Send>;
+        let exit = run_datagram_stage(
+            FailStream {
+                fail_read: false,
+                fail_write: false,
+            },
+            egress.as_mut(),
+        )
+        .await;
+        assert_eq!(exit, StageExit::EgressEnded);
+    }
+
+    /// The `Io` exit via a failing write: a valid reply frame arrives
+    /// but the channel's write half fails — the stage tears the
+    /// association down.
+    #[tokio::test]
+    async fn datagram_stage_write_failure_is_the_io_exit() {
+        let inner = {
+            let mut v = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53)))
+                .to_be_bytes()
+                .unwrap()
+                .to_vec();
+            v.extend_from_slice(b"payload");
+            Bytes::from(v)
+        };
+        let mut egress = Box::new(OneReplyEgress { reply: Some(inner) }) as Box<dyn Egress + Send>;
+        let exit = run_datagram_stage(
+            FailStream {
+                fail_read: false,
+                fail_write: true,
+            },
+            egress.as_mut(),
+        )
+        .await;
+        assert_eq!(exit, StageExit::Io);
+    }
+
+    /// The `Io` exit via a failing channel read — the stage tears the
+    /// association down (a stream error is never a silent end).
+    #[tokio::test]
+    async fn datagram_stage_read_failure_is_the_io_exit() {
+        let mut egress = Box::new(ParkedEgress) as Box<dyn Egress + Send>;
+        let exit = run_datagram_stage(
+            FailStream {
+                fail_read: true,
+                fail_write: false,
+            },
+            egress.as_mut(),
+        )
+        .await;
+        assert_eq!(exit, StageExit::Io);
+    }
+
+    /// The `EgressEnded` exit via a failing `egress.send`, at the
+    /// serve layer (the drop-witness treatment
+    /// `egress_factory_called_once_and_dropped_with_the_stage` gives
+    /// the happy path): one datagram in, the egress refuses it, the
+    /// stage exits — EOF reaches the client and the egress drops with
+    /// the stage (teardown-at-birth on the failure path).
+    #[tokio::test]
+    async fn udp_egress_send_failure_ends_the_association_and_drops_the_egress() {
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        let dropped = Arc::new(AtomicBool::new(false));
+        let dropped_for_factory = Arc::clone(&dropped);
+        let factory: EgressFactory = Arc::new(move || {
+            let dropped = Arc::clone(&dropped_for_factory);
+            Box::pin(async move {
+                struct FailingSendWitness {
+                    flag: Arc<AtomicBool>,
+                }
+                impl crate::egress::Egress for FailingSendWitness {
+                    fn send(
+                        &mut self,
+                        _addr: TargetAddr,
+                        _payload: Bytes,
+                    ) -> futures::future::BoxFuture<'_, Result<(), EgressError>>
+                    {
+                        Box::pin(
+                            async move { Err(EgressError::General("egress send failed".into())) },
+                        )
+                    }
+
+                    fn next_reply(
+                        &mut self,
+                    ) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>>
+                    {
+                        Box::pin(async move { Ok(std::future::pending::<Option<Bytes>>().await) })
+                    }
+                }
+                impl Drop for FailingSendWitness {
+                    fn drop(&mut self) {
+                        self.flag.store(true, Ordering::SeqCst);
+                    }
+                }
+                Ok(Box::new(FailingSendWitness { flag: dropped }) as Box<dyn Egress + Send>)
+            })
+        });
+
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let handle = tokio::spawn(serve_socks5_over(
+            server_end,
+            None,
+            Some(factory),
+            None,
+            false,
+        ));
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::UDPAssociate,
+            &TargetAddr::Domain("send.fail".into(), 1080),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x00, "the associate reply succeeds");
+
+        // One datagram in: the egress refuses it → the stage exits.
+        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 4], 53)));
+        client
+            .write_all(&datagram_frame(&target, b"doomed"))
+            .await
+            .unwrap();
+
+        // The association ends: EOF (no reply datagram follows).
+        let mut buf = [0u8; 16];
+        let outcome =
+            tokio::time::timeout(std::time::Duration::from_secs(5), client.read(&mut buf))
+                .await
+                .expect("read resolves");
+        match outcome {
+            Ok(0) | Err(_) => {}
+            Ok(n) => panic!("EOF expected after the egress send failure — got {n} bytes"),
+        }
+        drop(client);
+        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+            .await
+            .expect("session ends")
+            .expect("join")
+            .expect("the stage exited");
+        assert!(
+            dropped.load(Ordering::SeqCst),
+            "the stage exit drops the egress on the failure path too"
+        );
+    }
+
     /// CONNECT and ASSOCIATE coexist on one `serve_socks5_over` (the
     /// regression guard, extended): both commands serve their data
     /// planes over separate duplexes on the same seam set.
@@ -2163,6 +2455,182 @@ mod tests {
             .expect("join");
     }
 
+    /// The quiet wait's `Err(_)` arm (review 002 N-8): a channel read
+    /// that *errors* (I/O error, not EOF) during the quiet wait ends
+    /// the wait with a clean `Ok(())` — the accept future drops with
+    /// the session (the client-gone contract's error-shaped twin).
+    /// Driven directly on `bind_quiet_wait_and_serve` (a read error is
+    /// not producible over a duplex).
+    #[tokio::test]
+    async fn bind_quiet_wait_read_error_ends_the_wait_cleanly() {
+        use std::cell::Cell;
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        let dropped = Arc::new(AtomicBool::new(false));
+        struct FailReadStream {
+            polled: Cell<bool>,
+        }
+        impl tokio::io::AsyncRead for FailReadStream {
+            fn poll_read(
+                self: Pin<&mut Self>,
+                cx: &mut Context<'_>,
+                _buf: &mut tokio::io::ReadBuf<'_>,
+            ) -> Poll<std::io::Result<()>> {
+                // Park once (both branches get polled — the accept
+                // future starts, its owned state exists), then error
+                // on the wake: the read branch's error resolves after
+                // the accept branch has started.
+                if self.polled.get() {
+                    Poll::Ready(Err(std::io::Error::other("quiet-wait read failed")))
+                } else {
+                    self.polled.set(true);
+                    cx.waker().wake_by_ref();
+                    Poll::Pending
+                }
+            }
+        }
+        impl tokio::io::AsyncWrite for FailReadStream {
+            fn poll_write(
+                self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+                buf: &[u8],
+            ) -> Poll<std::io::Result<usize>> {
+                Poll::Ready(Ok(buf.len()))
+            }
+            fn poll_flush(
+                self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+            ) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+            fn poll_shutdown(
+                self: Pin<&mut Self>,
+                _cx: &mut Context<'_>,
+            ) -> Poll<std::io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+        }
+
+        struct OwnedAccept(Arc<AtomicBool>);
+        impl Drop for OwnedAccept {
+            fn drop(&mut self) {
+                self.0.store(true, Ordering::SeqCst);
+            }
+        }
+        let accept = {
+            let dropped = Arc::clone(&dropped);
+            BindAccept::new(SocketAddr::from(([127, 0, 0, 1], 2121)), async move {
+                let owner = OwnedAccept(dropped);
+                std::future::pending::<()>().await;
+                drop(owner);
+                unreachable!("pending never resolves")
+            })
+        };
+
+        let outcome = bind_quiet_wait_and_serve(
+            FailReadStream {
+                polled: Cell::new(false),
+            },
+            accept,
+        )
+        .await;
+        outcome.expect("an error-shaped quiet wait is not an error exit");
+        assert!(
+            dropped.load(Ordering::SeqCst),
+            "the accept future dropped with the wait (the client-gone contract, error arm)"
+        );
+    }
+
+    /// The BIND reply#2 Domain-peer fallback (review 002 N-8): a
+    /// foreign accept handle carrying a `TargetAddr::Domain(_, port)`
+    /// peer — the producer writes reply#2 with the unspecified
+    /// address on the peer's port (`0.0.0.0:port`) and serves the
+    /// data plane.
+    #[tokio::test]
+    async fn bind_domain_peer_falls_back_to_the_unspecified_address() {
+        let factory: BindAcceptFactory = Arc::new(move |_target: TargetAddr| {
+            Box::pin(async move {
+                Ok(BindAccept::new(
+                    SocketAddr::from(([127, 0, 0, 1], 2121)),
+                    async {
+                        let (mut echo_end, target_end) = tokio::io::duplex(64);
+                        tokio::spawn(async move {
+                            let mut buf = [0u8; 64];
+                            loop {
+                                match echo_end.read(&mut buf).await {
+                                    Ok(0) | Err(_) => return,
+                                    Ok(n) => {
+                                        if echo_end.write_all(&buf[..n]).await.is_err() {
+                                            return;
+                                        }
+                                    }
+                                }
+                            }
+                        });
+                        let (read_half, write_half) = tokio::io::split(target_end);
+                        Ok(BindTargetHandle {
+                            peer: TargetAddr::Domain("peer.example".into(), 44444),
+                            read: Box::new(read_half) as Box<dyn AsyncRead + Send + Sync + Unpin>,
+                            write: Box::new(write_half)
+                                as Box<dyn AsyncWrite + Send + Sync + Unpin>,
+                        })
+                    },
+                ))
+            })
+        });
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let handle = tokio::spawn(serve_socks5_over(
+            server_end,
+            None,
+            None,
+            Some(factory),
+            true,
+        ));
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::TCPBind,
+            &TargetAddr::Domain("domain.peer".into(), 2121),
+        )
+        .await;
+
+        // Reply#1 succeeds; reply#2 carries the Domain peer's fallback
+        // (the unspecified address, the peer's port).
+        let (code1, _addr) = read_reply(&mut client).await;
+        assert_eq!(code1, 0x00, "reply#1 succeeded");
+        let (code2, addr) = read_reply(&mut client).await;
+        assert_eq!(code2, 0x00, "reply#2 succeeded");
+        assert_eq!(
+            SocketAddr::from((
+                [addr[0], addr[1], addr[2], addr[3]],
+                u16::from_be_bytes([addr[4], addr[5]])
+            )),
+            SocketAddr::from(([0, 0, 0, 0], 44444)),
+            "the Domain peer fell back to 0.0.0.0:port"
+        );
+
+        // The data plane is live (the accept resolved with in-process
+        // halves — the parked probe half stays alive by forget).
+        client.write_all(b"ping").await.unwrap();
+        let mut buf = [0u8; 4];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            client.read_exact(&mut buf),
+        )
+        .await
+        .expect("the data plane echoes")
+        .expect("read the echo");
+        assert_eq!(&buf, b"ping");
+
+        drop(client);
+        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
+            .await
+            .expect("session ends")
+            .expect("join")
+            .expect("clean");
+    }
+
     /// The pump handler's shape: `accept_bi` → the same conversation,
     /// with the handler's `JoinHandle` tracking the data plane (R-02
     /// — the spawn shape is inline-await, not spawn-and-forget).
@@ -2232,6 +2700,15 @@ mod tests {
             Some("opener"),
             "the per-call opener identity is recorded"
         );
+
+        // The None-witness arm (the registration's optional witness
+        // seam): the session gate passes without recording anything.
+        let establisher = socks5_session_establisher(None);
+        let auth = AuthContext::anonymous(b"alk/socks5");
+        let establishment = establisher(serde_json::json!({}), auth)
+            .await
+            .expect("the session gate passes without a witness");
+        assert!(establishment.plan.is_none(), "no plan payload");
     }
 
     /// The command gate's resolved boolean rides the open-op