producer: the session-gate establisher + the command-dispatch pump + CONNECT

The crate's core task (implementation/core-producer-connect): the
producer half of src/producer.rs per producer.md, adapted to the
crate's seam types and the alkcall 0.8.0 establishment surface.

- The session-gate establisher (`socks5_session_establisher`):
  validates/records the opener identity (the CF-005/CF-006 seam,
  ADR-005) via the optional `identity_witness` probe (the alktunnels
  posture); returns `Establishment::default()` — no plan payload
  (the dial happens post-establishment; POC #1's wire note).
- The pump handler (`make_socks5_pump_handler`): `accept_bi()` →
  the forked typestate over the `BiStream` (`accept_no_auth →
  read_command`) → dispatch. CONNECT: the dial seam (`DialFn`,
  ADR-004) at the command-read interception point; success:
  `reply_success(sentinel)` then `pump_bidi` awaited inline (R-02 —
  the JoinHandle tracks the data-plane lifetime). Refusal:
  `reply_error` with the mapped RFC code — no phantom data plane.
- The ASSOCIATE/BIND arms land as RFC-faithful refusals
  (`CommandNotSupported`) — the pre-vertical posture (ADR-008); the
  dispatch skeleton + registration signature carry the seam
  parameters from day one so the associate/bind tasks swap the arms
  without touching the registration shape.
- `register_socks5_openable`: spec + establisher + pump via
  `register_openable_with_establisher` (`timeout: None` = the 10s
  default), scope-gated `socks5:open`. Missing callbacks = those
  commands refuse RFC-faithfully, never a panic.
- `serve_socks5_over<T>`: the substrate-free protocol entry (the
  POC shape), directly testable without channels; the builder
  (`Socks5Producer::builder()`) is the spec'd assembly shape.
- Tests: 11 in-module (substrate-free duplex-driven — round-trip +
  half-close teardown, refusal mapping per variant, missing-dial
  refusal, ASSOCIATE/BIND arms, dispatch guard, establisher witness,
  builder shape) + 5 harness integration (end-to-end CONNECT echo
  with the identity witness, dial refusal 0x05 with no phantom data
  plane, scope-class refusal 0x02, ASSOCIATE/BIND refusals with
  CONNECT still serving, repeated CONNECTs on one registration).
  The harness retargeted to the real producer (`RegisterMode::
  Producer`); the minimal stand-in is a harness-internal helper.

Verification: cargo test 87+19+37 green; clippy --all-targets -D
warnings clean; fmt clean; wasm check + wasm clippy clean; cargo
doc clean; --all-features green.

tests/harness.rs

@@ -421,6 +421,17 @@ pub enum RegisterMode {
     /// the given dial, with an identity-witnessing establisher (the
     /// CF-006 seam probe rides the same topology).
     MinimalConnect(DialFn),
+    /// The real producer registration
+    /// ([`alksocks::producer::register_socks5_openable`]): the
+    /// session-gate establisher + the command-dispatch pump handler
+    /// over the injected seams (dial required for CONNECT; the
+    /// egress/accept arms refuse RFC-faithfully until their tasks
+    /// land).
+    Producer {
+        dial: DialFn,
+        egress: Option<EgressFactory>,
+        accept: Option<BindAcceptFactory>,
+    },
     /// No socks5 openable — only the generic channel ops (the
     /// unknown-op / registry-gate probe's shape).
     NoSocks5,
@@ -618,6 +629,27 @@ pub async fn wire_with(
         )
         .expect("register socks5 openable (minimal CONNECT stand-in)");
     }
+    if let RegisterMode::Producer {
+        dial,
+        egress,
+        accept,
+    } = &mode
+    {
+        let core = ChannelCore::new(
+            producer_client.manager().clone(),
+            alkcall::channels::policy::default_policy(),
+        );
+        alksocks::producer::register_socks5_openable(
+            &core,
+            &producer_op_registry,
+            AuthContext::anonymous(b"alk/socks5"),
+            Some(Arc::clone(dial)),
+            egress.clone(),
+            accept.clone(),
+            Some(Arc::clone(&identity_witness)),
+        )
+        .expect("register socks5 openable (the real producer)");
+    }
 
     // --- consumer pieces (captured from the hook) -----------------------
     let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn");
@@ -634,11 +666,15 @@ pub async fn wire_with(
 }
 
 /// The default topology: the consumer's transport identity is the
-/// primary caller-identity path; the producer registers the minimal
-/// CONNECT stand-in over the given dial.
+/// primary caller-identity path; the producer registers the real
+/// CONNECT path over the given dial.
 pub async fn wire(dial: DialFn) -> Topology {
     wire_with(
-        RegisterMode::MinimalConnect(dial),
+        RegisterMode::Producer {
+            dial,
+            egress: None,
+            accept: None,
+        },
         Some(consumer_identity()),
         None,
         Arc::new(alkcall::core::auth::NoopIdentityProvider),
@@ -1087,7 +1123,11 @@ mod tests {
             resources: std::collections::HashMap::new(),
         };
         let topo = wire_with(
-            RegisterMode::MinimalConnect(echo_dial()),
+            RegisterMode::Producer {
+                dial: echo_dial(),
+                egress: None,
+                accept: None,
+            },
             Some(bare),
             None,
             Arc::new(alkcall::core::auth::NoopIdentityProvider),
@@ -1163,4 +1203,203 @@ mod tests {
             .expect("adopt");
         assert!(topo.consumer_manager.has_channel(channel_id));
     }
+
+    // -- the real-producer integration tests -----------------------------
+
+    /// Open a socks5 channel + adopt it + pump the raw RFC client
+    /// through it (the smoke test's opening shape, shared by the
+    /// producer integration tests).
+    async fn open_channel_and_pump(
+        topo: &Topology,
+    ) -> (
+        RawRfcClient<tokio::io::DuplexStream>,
+        tokio::task::JoinHandle<(u64, u64)>,
+    ) {
+        let response = topo
+            .consumer_call
+            .call(OP_SOCKS5_OPEN, Socks5Params::as_json())
+            .await;
+        let out = response.result.expect("the socks5 channel opens");
+        let channel_id = out
+            .get("channel_id")
+            .and_then(|v| v.as_u64())
+            .expect("the reply carries channel_id") as u32;
+        let (send, recv) = topo
+            .consumer_manager
+            .adopt_channel(channel_id, SOCKS5_ALPN, None)
+            .await
+            .expect("adopt the producer-allocated id");
+        let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024);
+        let pump = tokio::spawn(pump_bidi(
+            stream_self,
+            Box::new(recv) as Box<dyn AsyncRead + Send + Unpin>,
+            Box::new(send) as Box<dyn AsyncWrite + Send + Unpin>,
+        ));
+        (RawRfcClient::new(stream_peer), pump)
+    }
+
+    /// The real producer end-to-end (the smoke test through the
+    /// command-dispatch path, not the minimal stand-in): CONNECT
+    /// round-trip + echo + the identity witness (CF-006) + half-close
+    /// teardown (client EOF ends the session cleanly).
+    #[tokio::test]
+    async fn producer_connect_echo_witnesses_identity_and_tears_down_cleanly() {
+        let topo = wire(echo_dial()).await;
+        let (mut raw, pump) = open_channel_and_pump(&topo).await;
+
+        let code = tokio::time::timeout(
+            std::time::Duration::from_secs(10),
+            raw.connect(&TargetAddr::Domain("echo.test".into(), 443)),
+        )
+        .await
+        .expect("CONNECT round-trip timed out")
+        .expect("the raw client completes");
+        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+
+        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");
+
+        // The establisher witnessed the per-call opener identity.
+        assert_eq!(
+            topo.identity_witness.lock().await.as_deref(),
+            Some("consumer")
+        );
+
+        // Half-close teardown: client EOF ends the session; the
+        // channel pump completes (the handler tracked the data plane
+        // to the end — R-02).
+        drop(raw);
+        tokio::time::timeout(std::time::Duration::from_secs(10), pump)
+            .await
+            .expect("the data plane pump ends on client EOF")
+            .expect("join");
+    }
+
+    /// A refusing dial surfaces 0x05 in-stream with no phantom data
+    /// plane: the reply comes back refused; bytes sent after the
+    /// refusal are not echoed (nothing was dialed). The channel
+    /// stays healthy — a fresh open succeeds immediately.
+    #[tokio::test]
+    async fn producer_dial_refusal_is_0x05_and_no_phantom_data_plane() {
+        let topo = wire(refusing_dial(SocksDialError::Refused("target down".into()))).await;
+
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        let method = raw.greet().await.expect("greet");
+        assert_eq!(method, consts::SOCKS5_AUTH_METHOD_NONE);
+        raw.request(
+            Socks5Command::TCPConnect,
+            &TargetAddr::Domain("down.test".into(), 80),
+        )
+        .await
+        .expect("request written");
+        let (code, _addr) = raw.read_reply().await.expect("reply read");
+        assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED);
+
+        // No phantom data plane: nothing echoes (the refusal is the
+        // whole conversation; the data plane never existed).
+        drop(_pump);
+
+        // The channel topology is healthy: a fresh CONNECT through a
+        // working dial succeeds immediately after the refused one.
+        let topo2 = wire(echo_dial()).await;
+        let (mut raw2, _pump2) = open_channel_and_pump(&topo2).await;
+        let code = raw2
+            .connect(&TargetAddr::Domain("echo.test".into(), 443))
+            .await
+            .expect("fresh connect");
+        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+    }
+
+    /// The scope-class dial refusal (`NotAllowed`) surfaces 0x02
+    /// (the mapped code — the dial seam's policy refusal path).
+    #[tokio::test]
+    async fn producer_scope_class_refusal_is_0x02() {
+        let topo = wire(refusing_dial(SocksDialError::NotAllowed("policy".into()))).await;
+        let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+        raw.greet().await.expect("greet");
+        raw.request(
+            Socks5Command::TCPConnect,
+            &TargetAddr::Domain("policy.test".into(), 80),
+        )
+        .await
+        .expect("request written");
+        let (code, _addr) = raw.read_reply().await.expect("reply read");
+        assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_NOT_ALLOWED);
+    }
+
+    /// The ASSOCIATE/BIND dispatch arms refuse 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).
+    #[tokio::test]
+    async fn producer_associate_and_bind_refuse_0x07_connect_still_serves() {
+        let topo = wire(echo_dial()).await;
+
+        // ASSOCIATE refuses.
+        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);
+        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
+            .connect(&TargetAddr::Domain("echo.test".into(), 443))
+            .await
+            .expect("connect");
+        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+        raw.stream.write_all(b"still").await.expect("write");
+        let mut buf = [0u8; 5];
+        raw.stream.read_exact(&mut buf).await.expect("echo");
+        assert_eq!(&buf, b"still");
+    }
+
+    /// 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.
+    #[tokio::test]
+    async fn producer_serves_repeated_connects_on_one_registration() {
+        let topo = wire(echo_dial()).await;
+        for i in 0..3 {
+            let (mut raw, _pump) = open_channel_and_pump(&topo).await;
+            let code = raw
+                .connect(&TargetAddr::Domain(format!("t{i}.test"), 443))
+                .await
+                .expect("connect");
+            assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
+            raw.stream.write_all(b"ping").await.expect("write");
+            let mut buf = [0u8; 4];
+            raw.stream.read_exact(&mut buf).await.expect("echo");
+            assert_eq!(&buf, b"ping");
+            drop(raw);
+        }
+    }
 }