producer: the explicit bind-policy gate + the builder-register pin (review-002 C-3, D-2, S-1, N-2)

- C-3: Socks5ProducerBuilder.bind_policy(bool) — the substrate-free
  entry's explicit BIND gate, defaulting to the accept seam's
  presence; seam presence is never the sole gate
- D-2: RegisterMode::ProducerBuilder + an integration test wiring the
  full builder through .register(...) serving all three commands
- S-1: run_datagram_stage's doc names the hot-Ok(None) residual
- N-2: wire.md §UDP ASSOCIATE carries the inline-send backpressure line

Verified: cargo test (352 green), clippy/fmt/doc clean, wasm checks clean.

src/producer.rs

@@ -80,7 +80,10 @@
 //! boolean). The channel `BiStream` and a `TcpStream` are equally
 //! `T` — this is what makes the producer testable without channels
 //! (the differential-suite pattern) and reusable by the batch-2
-//! local backend (feature `local`, ADR-009).
+//! local backend (feature `local`, ADR-009). The builder's shape
+//! resolves that gate from its explicit `bind_policy` (defaulting to
+//! the accept seam's presence — never the sole gate; ADR-006's
+//! "listener configured" ≠ "listener permitted").
 //!
 //! ## Registration
 //!
@@ -346,7 +349,19 @@ where
 /// flow table); one relay loop, no spawn-per-datagram. A frame's
 /// egress `send` is awaited inline before the next select iteration —
 /// the stage interleaves egress sends with reply drains (the POC's
-/// measured shape; sufficient for relay postures).
+/// measured shape; sufficient for relay postures; a slow egress
+/// applies backpressure to the association, never to other channels —
+/// wire.md §UDP ASSOCIATE).
+///
+/// Residual posture (review 002 S-1, named decision): the `Ok(None)`
+/// tolerance's defense is bounded by `yield_now` only — a *misbehaving*
+/// egress returning `Ok(None)` at full speed degrades the stage into
+/// a yielding busy-wait (making no progress, unbounded by rate; the
+/// runtime survives, the stage never parks). A compliant egress parks
+/// (the readiness contract); a backoff/budget in the stage would
+/// penalize the compliant path, so the residual is bounded by the
+/// egress's own correctness and accepted-uncovered — named here, not
+/// accidental.
 async fn run_datagram_stage<T>(mut inner: T, egress: &mut (dyn Egress + Send)) -> StageExit
 where
     T: AsyncRead + AsyncWrite + Unpin,
@@ -615,11 +630,19 @@ pub fn register_socks5_openable(
 /// closes over substrate code behind the callbacks; the protocol
 /// layer never sees a socket type (ADR-004/ADR-008). Missing
 /// callbacks = those commands refuse RFC-faithfully (ADR-008).
+///
+/// The BIND command gate on the substrate-free entry
+/// ([`Socks5Producer::serve_socks5_over`]) is the builder's
+/// `bind_policy` — explicit via `.bind_policy(bool)`, defaulting to
+/// the accept seam's presence for backward-compat. The channels
+/// path's gate is the pump handler's, resolved from the per-call
+/// opener identity (ADR-006) — this policy never touches it.
 #[derive(Clone, Default)]
 pub struct Socks5Producer {
     dial: Option<DialFn>,
     egress: Option<EgressFactory>,
     accept: Option<BindAcceptFactory>,
+    bind_policy: bool,
 }
 
 impl Socks5Producer {
@@ -629,10 +652,12 @@ impl Socks5Producer {
 
     /// The substrate-free protocol entry (the same shape
     /// [`serve_socks5_over`] pins) over the builder's seams. The
-    /// command gate is resolved from the builder's `accept` presence:
-    /// a producer that wired the accept seam serves BIND to whatever
-    /// calls this entry directly (the per-caller gate is the pump
-    /// handler's, resolved from the open-op identity — ADR-006).
+    /// command gate is the builder's `bind_policy` — an explicit
+    /// `.bind_policy(bool)` whenever the local policy must speak
+    /// ("factory wired but gate closed" is expressible here);
+    /// unwired, it defaults to the accept seam's presence for
+    /// backward-compat. The per-caller gate is the pump handler's,
+    /// resolved from the open-op identity — ADR-006.
     pub async fn serve_socks5_over<T>(&self, stream: T) -> Result<(), SocksError>
     where
         T: AsyncRead + AsyncWrite + Unpin,
@@ -642,7 +667,7 @@ impl Socks5Producer {
             self.dial.clone(),
             self.egress.clone(),
             self.accept.clone(),
-            self.accept.is_some(),
+            self.bind_policy,
         )
         .await
     }
@@ -675,6 +700,7 @@ pub struct Socks5ProducerBuilder {
     dial: Option<DialFn>,
     egress: Option<EgressFactory>,
     accept: Option<BindAcceptFactory>,
+    bind_policy: Option<bool>,
 }
 
 impl Socks5ProducerBuilder {
@@ -696,8 +722,21 @@ impl Socks5ProducerBuilder {
         self
     }
 
+    /// The substrate-free entry's BIND command gate (ADR-006's
+    /// posture — "listener configured" ≠ "listener permitted"): the
+    /// resolved `can_bind` for [`Socks5Producer::serve_socks5_over`].
+    /// Set it when the local policy must close the gate while the
+    /// accept seam stays wired. Unwired, the gate defaults to the
+    /// accept seam's presence (the backward-compat reading — the
+    /// pump handler's identity-derived gate is unaffected).
+    pub fn bind_policy(mut self, can_bind: bool) -> Self {
+        self.bind_policy = Some(can_bind);
+        self
+    }
+
     pub fn build(self) -> Socks5Producer {
         Socks5Producer {
+            bind_policy: self.bind_policy.unwrap_or(self.accept.is_some()),
             dial: self.dial,
             egress: self.egress,
             accept: self.accept,
@@ -2249,6 +2288,193 @@ mod tests {
         result.expect("clean");
     }
 
+    /// The gate policy (review 002 C-3): the builder's `bind_policy`
+    /// is the substrate-free entry's BIND gate — "factory wired but
+    /// gate closed" is expressible (`bind_policy(false)` with the
+    /// accept seam still wired refuses BIND 0x02, no factory call —
+    /// the gate precedes the seam) while CONNECT and ASSOCIATE still
+    /// serve the same producer. Seam presence is never the sole gate.
+    #[tokio::test]
+    async fn builder_bind_policy_false_refuses_bind_while_connect_associate_serve() {
+        let requested = Arc::new(tokio::sync::Mutex::new(None));
+        let producer = Socks5Producer::builder()
+            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
+            .egress(echo_egress_factory())
+            .accept(bind_accept_factory(Arc::clone(&requested)))
+            .bind_policy(false)
+            .build();
+
+        // BIND with the factory wired and the gate closed: 0x02, and
+        // the accept seam is never called (no phantom listener — the
+        // gate precedes the seam, the ADR-006 shape).
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::TCPBind,
+            &TargetAddr::Domain("policy.closed".into(), 2121),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(
+            code, 0x02,
+            "the explicit policy refuses ConnectionNotAllowed"
+        );
+        assert!(
+            requested.lock().await.is_none(),
+            "the closed gate precedes the accept seam — no phantom listener"
+        );
+        drop(client);
+        let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
+            .await
+            .expect("session ends")
+            .expect("join")
+            .expect_err("the refusal surfaces in-stream");
+        assert!(matches!(
+            err,
+            SocksError::ServerError(SocksServerError::AuthenticationRejected)
+        ));
+
+        // CONNECT still serves the same producer (egress unaffected).
+        let producer = Socks5Producer::builder()
+            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
+            .egress(echo_egress_factory())
+            .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None))))
+            .bind_policy(false)
+            .build();
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::TCPConnect,
+            &TargetAddr::Domain("policy.connect".into(), 443),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x00, "CONNECT is unaffected by the bind policy");
+        client.write_all(b"ping").await.unwrap();
+        let mut buf = [0u8; 4];
+        client.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"ping");
+        drop(client);
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
+            .await
+            .expect("session ends")
+            .expect("join");
+        result.expect("clean");
+
+        // ASSOCIATE still serves the same producer.
+        let producer = Socks5Producer::builder()
+            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
+            .egress(echo_egress_factory())
+            .accept(bind_accept_factory(Arc::new(tokio::sync::Mutex::new(None))))
+            .bind_policy(false)
+            .build();
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::UDPAssociate,
+            &TargetAddr::Domain("policy.udp".into(), 1080),
+        )
+        .await;
+        let (code, addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x00, "ASSOCIATE is unaffected by the bind policy");
+        assert_eq!(
+            (addr[0], addr[1], addr[2], addr[3], addr[4], addr[5]),
+            (127, 0, 0, 1, 0, 0),
+            "the reply carries the sentinel BND.ADDR"
+        );
+        let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 8], 53)));
+        client
+            .write_all(&datagram_frame(&target, b"policy"))
+            .await
+            .unwrap();
+        let mut buf = vec![0u8; 2 + 7 + 6];
+        client.read_exact(&mut buf).await.expect("echo");
+        let (_, echoed) = decode_datagram(&buf);
+        assert_eq!(&echoed[..], b"policy");
+        drop(client);
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
+            .await
+            .expect("session ends")
+            .expect("join");
+        result.expect("clean");
+    }
+
+    /// The backward-compat default (review 002 C-3): an unwired
+    /// `bind_policy` resolves the gate from the accept seam's
+    /// presence — a factory-wired producer serves BIND on the
+    /// substrate-free entry (the landed tests' reading preserved),
+    /// while a producer without the factory still refuses. The
+    /// explicit policy always wins when wired.
+    #[tokio::test]
+    async fn builder_default_bind_policy_follows_accept_presence() {
+        // Accept wired, policy unwired → BIND serves (the default).
+        let requested = Arc::new(tokio::sync::Mutex::new(None));
+        let producer = Socks5Producer::builder()
+            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
+            .accept(bind_accept_factory(Arc::clone(&requested)))
+            .build();
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::TCPBind,
+            &TargetAddr::Domain("default.open".into(), 2121),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x00, "the presence default serves BIND");
+        assert!(
+            requested.lock().await.is_some(),
+            "the accept seam ran (the factory is wired and the gate open)"
+        );
+        drop(client);
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
+            .await
+            .expect("session ends")
+            .expect("join");
+        result.expect("clean");
+
+        // Accept unwired, policy unwired → BIND refuses (the gate's
+        // no-grant answer — no seam to run even if the gate were
+        // open).
+        let producer = Socks5Producer::builder()
+            .dial(echo_dial(Arc::new(tokio::sync::Mutex::new(Vec::new()))))
+            .build();
+        let (client_end, server_end) = tokio::io::duplex(64 * 1024);
+        let serve = tokio::spawn(async move { producer.serve_socks5_over(server_end).await });
+        let mut client = client_end;
+        greet(&mut client).await;
+        write_request(
+            &mut client,
+            &Socks5Command::TCPBind,
+            &TargetAddr::Domain("default.closed".into(), 2121),
+        )
+        .await;
+        let (code, _addr) = read_reply(&mut client).await;
+        assert_eq!(code, 0x02, "no factory + no policy = the closed default");
+        drop(client);
+        let err = tokio::time::timeout(std::time::Duration::from_secs(5), serve)
+            .await
+            .expect("session ends")
+            .expect("join")
+            .expect_err("the refusal surfaces in-stream");
+        assert!(matches!(
+            err,
+            SocksError::ServerError(SocksServerError::AuthenticationRejected)
+        ));
+    }
+
     /// The sentinel reply address (the in-band BND.ADDR posture,
     /// ADR-003): `127.0.0.1:0` — the RFC-compatibility surface.
     #[test]