producer: the BIND phase model + the socks5:bind command gate (ADR-006/ADR-007)

The TCP BIND dispatch arm replaces its refusal with the full phase
model (POC #5-validated scope), plus the command gate at dispatch:

- The command gate: TCP BIND requires socks5:bind from the per-call
  opener identity (CF-005/CF-006) — the pump handler resolves it
  from the open-op AuthContext; serve_socks5_over<T> takes the
  resolved can_bind boolean explicitly. A scope-less identity gets
  the RFC-faithful ConnectionNotAllowed (0x02) with no factory call
  (the gate precedes the seam) while CONNECT still serves it.
- The phase model (wire.md §BIND): the BindAcceptFactory at the
  command-read interception point → reply#1 carrying BND.ADDR (the
  factory's bind_addr) → the quiet wait (single-shot select!, exactly
  two exits: the one accept, or the control stream's EOF/garbage/
  error; any byte = protocol garbage → session ends, no data plane;
  the accept future's drop IS the client-gone teardown) → reply#2
  carrying the peer address (via the fork's published new_reply) →
  pump_bidi inline (the CONNECT shape, R-02).
- Two-stage RFC-faithful failure mapping: factory failure → reply#1's
  mapped code; accept-stage failure → reply#2's mapped code, no data
  plane. No phantom listeners, no phantom data planes.
- The no-factory default is the wrapper-aware in-band variant
  (ADR-007 variant 3, documented): sentinel reply#1 + sentinel reply#2
  (0.0.0.0:0), the full phase model preserved — the channel is the
  connection.
- The accept wait lives in the teardown-governed pump phase, never
  under the establishment bound (ADR-007).

Tests: 16 unit tests (raw RFC client over duplex) + 7 harness
integration tests over channels — happy path, the gate (0x02 while
CONNECT serves, both scopes), quiet-wait garbage, client-gone
teardown (owner-flag witness), both failure stages over channels,
no-factory variant 3, and the three-command regression guard (CONNECT
serves, BIND serves under its gate, ASSOCIATE still refuses).

Verification: cargo test green (96 lib + 26 harness + 37
differential); clippy -D warnings clean (all targets, native +
wasm32-unknown-unknown); cargo fmt clean; wasm check clean.

Task: tasks/implementation/core-producer-bind.md (status → completed)

tests/harness.rs

@@ -90,6 +90,13 @@ pub fn both_scopes_identity() -> Identity {
     }
 }
 
+/// The BIND-capable identity the BIND integration tests open with:
+/// both scopes (the open gate needs `socks5:open`, the command gate
+/// needs `socks5:bind` — a session needs both grants explicitly).
+pub fn bind_scoped_identity() -> Identity {
+    both_scopes_identity()
+}
+
 /// The scoped token — resolves to [`consumer_identity`] via
 /// [`TestIdProvider`] (the hub-forwarding path's shape).
 pub const TEST_AUTH_TOKEN: &str = "test-token-with-socks5-open-scope";
@@ -789,6 +796,13 @@ mod tests {
     fn assert_send_sync<T: Send + Sync>() {}
     fn assert_send<T: Send>() {}
 
+    fn init_trace() {
+        let _ = tracing_subscriber::fmt()
+            .with_max_level(tracing::level_filters::LevelFilter::TRACE)
+            .with_test_writer()
+            .try_init();
+    }
+
     #[test]
     fn identities_carry_the_intended_scopes() {
         let open = consumer_identity();
@@ -1335,12 +1349,13 @@ mod tests {
         assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED);
     }
 
-    /// The ASSOCIATE/BIND dispatch arms refuse RFC-faithfully
+    /// The ASSOCIATE dispatch arm refuses RFC-faithfully
     /// (`CommandNotSupported` 0x07 — the pre-vertical posture; the
-    /// producer-associate/bind tasks replace them) while CONNECT
-    /// still serves on the same registration (the regression guard).
+    /// producer-associate task replaces it) while CONNECT still
+    /// serves on the same registration. BIND now serves under the
+    /// gate — its integration tests live below.
     #[tokio::test]
-    async fn producer_associate_and_bind_refuse_0x07_connect_still_serves() {
+    async fn producer_associate_refuses_0x07_connect_still_serves() {
         let topo = wire(echo_dial()).await;
 
         // ASSOCIATE refuses.
@@ -1356,19 +1371,6 @@ mod tests {
         assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED);
         drop(raw);
 
-        // BIND refuses.
-        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
-        raw.greet().await.expect("greet");
-        raw.request(
-            Socks5Command::TCPBind,
-            &TargetAddr::Domain("bind.test".into(), 2121),
-        )
-        .await
-        .expect("request written");
-        let (code, _addr) = raw.read_reply().await.expect("reply read");
-        assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED);
-        drop(raw);
-
         // CONNECT still serves the same registration.
         let (mut raw, _pump) = open_channel_and_pump(&topo).await;
         let code = raw
@@ -1382,6 +1384,415 @@ mod tests {
         assert_eq!(&buf, b"still");
     }
 
+    /// BIND over channels (the phase model end to end): the accept
+    /// factory resolves at command dispatch, reply#1 carries the
+    /// advertised BND.ADDR, reply#2 carries the accepted peer, and
+    /// the data plane two-pumps raw through the channel (echo both
+    /// directions).
+    #[tokio::test]
+    async fn producer_bind_round_trip_over_channels() {
+        init_trace();
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(echo_accept()),
+            },
+            Some(bind_scoped_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        let (mut raw, pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+
+        // Reply#1: the advertised BND.ADDR (the harness's static
+        // in-process address).
+        let (code1, addr1) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED, "reply#1 success");
+        assert_eq!(
+            u16::from_be_bytes([addr1[addr1.len() - 2], addr1[addr1.len() - 1]]),
+            ECHO_BIND_PORT,
+            "reply#1 carries the factory's advertised port"
+        );
+
+        // Reply#2: the accepted peer (the harness's fixed stand-in).
+        let (code2, addr2) = raw.read_reply().await.expect("reply#2");
+        assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED, "reply#2 success");
+        let peer_port = u16::from_be_bytes([addr2[addr2.len() - 2], addr2[addr2.len() - 1]]);
+        assert_eq!(peer_port, 44444, "reply#2 carries the accepted peer");
+
+        // Phase 2: raw pass-through through the channel, both
+        // directions.
+        raw.stream.write_all(b"ping").await.expect("write");
+        let mut buf = [0u8; 4];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read_exact(&mut buf),
+        )
+        .await
+        .expect("echo timed out")
+        .expect("read the echo");
+        assert_eq!(&buf, b"ping");
+        raw.stream.write_all(b"pong").await.expect("write");
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read_exact(&mut buf),
+        )
+        .await
+        .expect("echo timed out")
+        .expect("read the echo");
+        assert_eq!(&buf, b"pong");
+
+        // Half-close teardown ends the session cleanly.
+        drop(raw);
+        tokio::time::timeout(std::time::Duration::from_secs(10), pump)
+            .await
+            .expect("the data plane pump ends on client EOF")
+            .expect("join");
+    }
+
+    /// The command gate over channels (ADR-006; POC #5 test 4): an
+    /// identity holding only `socks5:open` gets `ConnectionNotAllowed`
+    /// (0x02) on BIND — while CONNECT still serves the same identity
+    /// (the gate distinguishes ingress from egress).
+    #[tokio::test]
+    async fn producer_bind_command_gate_refuses_0x02_connect_unaffected() {
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(echo_accept()),
+            },
+            Some(consumer_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        // BIND refused, RFC-faithfully — no phantom listener (the
+        // accept seam never ran; the session ends).
+        let (mut raw, pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("gate.test".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code, _addr) = raw.read_reply().await.expect("reply read");
+        assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED);
+        drop(raw);
+        drop(pump);
+
+        // CONNECT still works for the same identity (egress is not
+        // implied by the BIND gate's absence — separate scopes).
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        let code = raw
+            .connect(&TargetAddr::Domain("echo.test".into(), 443))
+            .await
+            .expect("connect still allowed");
+        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+        raw.stream.write_all(b"egress").await.expect("write");
+        let mut buf = [0u8; 6];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read_exact(&mut buf),
+        )
+        .await
+        .expect("echo timed out")
+        .expect("read the echo");
+        assert_eq!(&buf, b"egress");
+    }
+
+    /// The BIND-scope-only identity over channels: `socks5:bind`
+    /// alone serves BIND (the open gate is `socks5:open` — the
+    /// registration's identity must carry it; the command gate only
+    /// adds `socks5:bind` on top of a granted session). The both-
+    /// scopes identity is the deployment shape that grants both
+    /// explicitly (ADR-006: never implied).
+    #[tokio::test]
+    async fn producer_bind_serves_a_both_scopes_identity() {
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(echo_accept()),
+            },
+            Some(both_scopes_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code1, _addr) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED);
+        let (code2, _addr) = raw.read_reply().await.expect("reply#2");
+        assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED);
+        raw.stream.write_all(b"scoped").await.expect("write");
+        let mut buf = [0u8; 6];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read_exact(&mut buf),
+        )
+        .await
+        .expect("echo timed out")
+        .expect("read the echo");
+        assert_eq!(&buf, b"scoped");
+    }
+
+    /// Quiet-wait garbage over channels: a control byte after
+    /// reply#1 ends the session with no data plane — the raw client
+    /// reads EOF (no reply#2, no echo), and the channel pump
+    /// completes.
+    #[tokio::test]
+    async fn producer_bind_quiet_wait_garbage_over_channels() {
+        init_trace();
+        // The accept future parks (never resolves) so the garbage
+        // branch is deterministic.
+        let parked: BindAcceptFactory = Arc::new(move |_target: TargetAddr| {
+            Box::pin(async move {
+                let bind_addr =
+                    SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), ECHO_BIND_PORT);
+                Ok(BindAccept::new(
+                    bind_addr,
+                    std::future::pending::<Result<BindTargetHandle, BindAcceptError>>(),
+                ))
+            })
+        });
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(parked),
+            },
+            Some(bind_scoped_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        let (mut raw, pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("quiet.test".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code1, _addr) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED);
+
+        // Protocol garbage in the quiet wait.
+        raw.stream.write_all(&[0x05, 0x02]).await.expect("garbage");
+
+        // The session ends: EOF (no reply#2, no echo).
+        let mut buf = [0u8; 10];
+        let outcome = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read(&mut buf),
+        )
+        .await
+        .expect("read resolves");
+        match outcome {
+            Ok(0) | Err(_) => {}
+            Ok(n) => panic!("no reply#2 after garbage — got {n} bytes"),
+        }
+        // The client vanishes (EOF both ways through the raw pipe);
+        // the session pump ends (R-02 — the handler tracked the data
+        // plane to its end).
+        drop(raw);
+        tokio::time::timeout(std::time::Duration::from_secs(10), pump)
+            .await
+            .expect("the session pump ends on the garbage refusal")
+            .expect("join");
+    }
+
+    /// The three-command regression guard over channels (POC #5 test
+    /// 7): one registration, all three commands — CONNECT serves,
+    /// BIND serves under its gate, ASSOCIATE refuses (its task
+    /// pending). Fresh channel per session.
+    #[tokio::test]
+    async fn producer_three_command_regression_guard() {
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(echo_accept()),
+            },
+            Some(both_scopes_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        // CONNECT serves.
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        let code = raw
+            .connect(&TargetAddr::Domain("echo.test".into(), 443))
+            .await
+            .expect("connect");
+        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+        raw.stream.write_all(b"c").await.expect("write");
+        let mut buf = [0u8; 1];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.stream.read_exact(&mut buf),
+        )
+        .await
+        .expect("echo timed out")
+        .expect("read the echo");
+        assert_eq!(&buf, b"c");
+        drop(raw);
+
+        // BIND serves under the gate.
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("bind.test".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code1, _addr) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED);
+        let (code2, _addr) = raw.read_reply().await.expect("reply#2");
+        assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED);
+        drop(raw);
+
+        // ASSOCIATE still refuses (the producer-associate task is
+        // pending).
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::UDPAssociate,
+            &TargetAddr::Domain("relay.test".into(), 1080),
+        )
+        .await
+        .expect("request written");
+        let (code, _addr) = raw.read_reply().await.expect("reply read");
+        assert_eq!(code, consts::SOCKS5_REPLY_COMMAND_NOT_SUPPORTED);
+    }
+
+    /// BIND two-stage failure mapping over channels: factory failure
+    /// → reply#1's mapped code; accept-stage failure → reply#2's
+    /// mapped code, no data plane (POC #5 test 6's shape).
+    #[tokio::test]
+    async fn producer_bind_failure_mapping_over_channels() {
+        use alksocks::egress::BindAcceptFactoryError;
+
+        // Factory failure → reply#1 ConnectionRefused (0x05).
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(refusing_accept_factory(BindAcceptFactoryError::Refused(
+                    "no listener".into(),
+                ))),
+            },
+            Some(bind_scoped_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("factory.fail".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code, _addr) = raw.read_reply().await.expect("reply read");
+        assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED);
+        drop(raw);
+
+        // Accept-stage failure → reply#2 ConnectionNotAllowed (0x02).
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: Some(refusing_accept(BindAcceptError::NotAllowed(
+                    "policy".into(),
+                ))),
+            },
+            Some(bind_scoped_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("accept.fail".into(), ECHO_BIND_PORT),
+        )
+        .await
+        .expect("request written");
+        let (code1, _addr) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(
+            code1,
+            consts::SOCKS5_REPLY_SUCCEEDED,
+            "the factory resolved"
+        );
+        let (code2, _addr) = raw.read_reply().await.expect("reply#2");
+        assert_eq!(code2, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED);
+    }
+
+    /// The no-factory default over channels (ADR-007 variant 3,
+    /// wrapper-aware): sentinel reply#1 + sentinel reply#2, the full
+    /// phase model preserved (the quiet wait included).
+    #[tokio::test]
+    async fn producer_bind_no_factory_serves_variant_3_over_channels() {
+        let topo = wire_with(
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: None,
+            },
+            Some(bind_scoped_identity()),
+            None,
+            Arc::new(alkcall::core::auth::NoopIdentityProvider),
+        )
+        .await;
+
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPBind,
+            &TargetAddr::Domain("variant3.test".into(), 2121),
+        )
+        .await
+        .expect("request written");
+        let (code1, addr1) = raw.read_reply().await.expect("reply#1");
+        assert_eq!(code1, consts::SOCKS5_REPLY_SUCCEEDED);
+        assert!(
+            addr1.iter().all(|b| *b == 0),
+            "reply#1 carries the wrapper-aware sentinel"
+        );
+        let (code2, addr2) = raw.read_reply().await.expect("reply#2");
+        assert_eq!(code2, consts::SOCKS5_REPLY_SUCCEEDED);
+        assert!(
+            addr2.iter().all(|b| *b == 0),
+            "reply#2 carries the wrapper-aware sentinel peer"
+        );
+    }
+
     /// Multiple sequential CONNECTs ride one registration: fresh
     /// channel per session (one channel per SOCKS5 session,
     /// convention 10), each dial seam call gets its parsed target.