//! Test harness: the producer↔consumer wiring over a
//! `tokio::io::duplex` carrying the channels 8-byte chunk format.
#![allow(dead_code)]
//!
//! Topology (the alktunnels harness shape, adapted to this crate's
//! seams; the alkcall 0.8.0 establishment surface under test; this
//! is the crate's first consumption of it):
//!
//! - **producer** (connect side of the channels connection; it runs
//!   `ChannelClient::from_connection_with_serving`): serves the socks5
//!   open op + the generic channel ops on its serving registry. Its
//!   manager allocates the channel IDs (`ChannelSide::Connect`, whose
//!   serving side allocates; the initiator adopts).
//! - **consumer** (accept side): the `ChannelsAdapter` + capturing
//!   install hook. The hook captures the consumer's channel-0
//!   `CallConnection` (for calling the producer's open op) and the
//!   consumer's `ChannelManager` (for adopting the producer-
//!   allocated IDs).
//!
//! The POC #1 wiring requirement is load-bearing here
//! (producer.md §Harness wiring requirement): the accept-side
//! install hook must call `channel0_conn.set_identity(resolved)`
//! before building the `CallConnection`; without it, scope-gated
//! opens fail closed with `FORBIDDEN "authentication required"`
//! (correct behavior, implicit wiring). The harness makes the
//! requirement explicit and tested.
//!
//! Everything is in-process: no sockets, no `local` feature. The
//! harness is dev-only (`tests/`), so it stays out of the wasm build.
//! The `allow(dead_code)` rides the alktunnels posture: each test
//! binary compiles the harness and uses a different subset.

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;

use bytes::Bytes;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
use alkcall::channels::client::{ChannelClient, ServingConfig};
use alkcall::channels::manager::ChannelManager;
use alkcall::channels::operations::{ChannelCore, ChannelOperations};
use alkcall::channels::pump::pump_bidi;
use alkcall::core::auth::{AuthContext, AuthToken, Identity, IdentityProvider};
use alkcall::core::types::Connection as CoreConnection;
use alkcall::protocol::connection::{split_single_stream, CallConnection};
use alkcall::protocol::dispatch::Dispatcher;
use alkcall::registry::registration::OperationRegistry;

use alksocks::egress::{
    BindAccept, BindAcceptError, BindAcceptFactory, BindTargetHandle, DialFn, EgressError,
    EgressFactory, SocksDialError,
};
use alksocks::params::{
    socks5_open_spec, OP_SOCKS5_OPEN, SOCKS5_ALPN, SOCKS5_BIND_SCOPE, SOCKS5_OPEN_SCOPE,
};
use alksocks::rfc1928::consts;
use alksocks::rfc1928::server::{new_reply, Socks5ServerProtocol, SocksServerError};
use alksocks::rfc1928::util::target_addr::TargetAddr;
use alksocks::rfc1928::{ReplyError, Socks5Command, SocksError};
use alksocks::Socks5Params;

// --- Shared helpers (the single-sourced shapes; review 002 N-5) -----------
//
// The dial/egress/accept stand-ins and the raw RFC 1928 client trio
// live in `tests/common/mod.rs`, the same file the lib test crate
// includes via `#[path]` (unit tests in `src/` cannot import
// `tests/harness.rs`, but they can share this module). The harness
// declares the module (path-included from `tests/`) and re-exports
// the shapes its tests consume so the per-test `use harness::{...}`
// surface is unchanged.

#[path = "common/mod.rs"]
#[allow(unused_imports)]
pub mod common;

#[allow(unused_imports)]
pub use common::{
    datagram_frame as udp_frame, decode_datagram as udp_unframe, doubled_echo_egress_factory,
    echo_dial, echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress,
    parked_accept_factory, raw_greet, raw_read_reply, raw_request, read_reply_v4, refusing_dial,
    refusing_egress, write_request, QueueEgress,
};

// --- Identities -----------------------------------------------------------

/// The consumer identity the transport carries: `socks5:open` only
/// (the CONNECT/ASSOCIATE grant; no `socks5:bind`).
pub fn consumer_identity() -> Identity {
    Identity {
        id: "consumer".to_string(),
        scopes: vec![SOCKS5_OPEN_SCOPE.to_string()],
        resources: HashMap::new(),
    }
}

/// The BIND-scope identity: `socks5:bind` ONLY; no `socks5:open`.
/// The command-gate probe's identity.
pub fn bind_identity() -> Identity {
    Identity {
        id: "consumer-bind".to_string(),
        scopes: vec![SOCKS5_BIND_SCOPE.to_string()],
        resources: HashMap::new(),
    }
}

/// Both scopes: `socks5:open` AND `socks5:bind` (a deployment
/// granting both grants both explicitly; never implied, ADR-006).
pub fn both_scopes_identity() -> Identity {
    Identity {
        id: "consumer-both".to_string(),
        scopes: vec![SOCKS5_OPEN_SCOPE.to_string(), SOCKS5_BIND_SCOPE.to_string()],
        resources: HashMap::new(),
    }
}

/// 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; it resolves to [`consumer_identity`] via
/// [`TestIdProvider`] (the hub-forwarding path's shape).
pub const TEST_AUTH_TOKEN: &str = "test-token-with-socks5-open-scope";

/// Resolve the scoped token to the consumer identity; anything else
/// resolves to none (falling through to the transport identity per
/// the CF-005 precedence order).
pub struct TestIdProvider;

impl IdentityProvider for TestIdProvider {
    fn resolve_from_fingerprint(&self, _: &str) -> Option<Identity> {
        None
    }
    fn resolve_from_token(&self, token: &AuthToken) -> Option<Identity> {
        if token.raw == TEST_AUTH_TOKEN.as_bytes() {
            Some(consumer_identity())
        } else {
            None
        }
    }
}

// --- In-process seam closures ---------------------------------------------
//
// The dial/egress/accept stand-ins are single-sourced in
// `tests/common/mod.rs` (declared + re-exported above): `echo_dial`,
// `refusing_dial`, `echo_egress` (the queue egress), and
// `parked_accept_factory`. The shapes below are the ones deliberately
// different per suite: the accept factories (echo + refusing) and the
// producer-side registrations.

/// The advertised BND.ADDR of [`echo_accept`] (in-process, no real
/// listener; the address is the reply#1 carrier only).
pub const ECHO_BIND_PORT: u16 = 2121;

/// An echo accept factory: reply#1 advertises a static in-process
/// address; the accept future resolves immediately with in-process
/// duplex halves (the POC #5 shape). The future owns its resources
/// end-to-end; the peer address is a fixed documented stand-in.
pub fn echo_accept() -> 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);
            let accept_future = async move {
                // The accepted data plane is an in-process echo pipe:
                // the producer's pump gets the `target_end` halves; a
                // spawned task echoes the `echo_end` (the "inbound
                // server" the BIND client's bytes round-trip through).
                let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024);
                tokio::spawn(async move {
                    loop {
                        let mut buf = vec![0u8; 4096];
                        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::Ip(SocketAddr::from(([10, 0, 0, 7], 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>,
                })
            };
            Ok(BindAccept::new(bind_addr, accept_future))
        })
    })
}

/// A failing accept factory (the BIND-refusal probe, pre-reply#1 →
/// reply#1's code). The variant is reconstructed per call (the seam
/// errors are not `Clone`).
pub fn refusing_accept_factory(err: alksocks::egress::BindAcceptFactoryError) -> BindAcceptFactory {
    Arc::new(move |_target: TargetAddr| {
        use alksocks::egress::BindAcceptFactoryError as E;
        let err = match &err {
            E::Refused(m) => E::Refused(m.clone()),
            E::NotAllowed(m) => E::NotAllowed(m.clone()),
            E::Shortage(m) => E::Shortage(m.clone()),
            E::General(m) => E::General(m.clone()),
        };
        Box::pin(async move { Err(err) })
    })
}

/// A failing accept future (the BIND-refusal probe, post-reply#1 →
/// reply#2's code): the factory resolves (reply#1 succeeds with the
/// advertised BND.ADDR), then the accept future refuses. The variant
/// is reconstructed per call (the seam errors are not `Clone`).
pub fn refusing_accept(err: BindAcceptError) -> BindAcceptFactory {
    Arc::new(move |_target: TargetAddr| {
        use BindAcceptError as E;
        let err = match &err {
            E::Refused(m) => E::Refused(m.clone()),
            E::NotAllowed(m) => E::NotAllowed(m.clone()),
            E::General(m) => E::General(m.clone()),
        };
        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, async move { Err(err) }))
        })
    })
}

// --- The producer registration helper --------------------------------------

/// What the establisher's per-call auth carried as the opener
/// identity id (the CF-006 probe; `None` until an open runs).
pub type IdentityWitness = Arc<tokio::sync::Mutex<Option<String>>>;

/// The session-gate establisher: records the opener identity (the
/// CF-006 seam the harness witnesses); returns
/// `Establishment::default()`, since no plan is needed (the dial is
/// in-stream, in the pump handler). A scope refusal is the registry
/// ACL gate's job (it runs before this ever runs; ADR-006).
pub fn socks5_session_establisher(
    witness: IdentityWitness,
) -> alkcall::channels::operations::OpenEstablisher {
    Arc::new(move |_input: serde_json::Value, auth: AuthContext| {
        let witness = Arc::clone(&witness);
        Box::pin(async move {
            *witness.lock().await = auth.identity.as_ref().map(|i| i.id.clone());
            Ok(alkcall::channels::operations::Establishment::default())
        })
    })
}

/// The smoke-test's minimal CONNECT producer path (the POC #1
/// `serve_socks5_over` shape, inline): negotiate no-auth → read the
/// command → dial via the injected closure → reply → `pump_bidi`
/// inline (R-02: the handler's `JoinHandle` tracks the data-plane
/// lifetime; early return = teardown-at-birth). The producer task
/// replaces this with the real command-dispatch pump; the harness
/// keeps the smoke test producer-independent.
async fn serve_connect_minimal<T>(stream: T, dial: DialFn) -> Result<(), SocksError>
where
    T: AsyncRead + AsyncWrite + Unpin,
{
    let (proto, cmd, target) = Socks5ServerProtocol::accept_no_auth(stream)
        .await
        .map_err(SocksError::ServerError)?
        .read_command()
        .await
        .map_err(SocksError::ServerError)?;

    if cmd != Socks5Command::TCPConnect {
        proto
            .reply_error(&ReplyError::CommandNotSupported)
            .await
            .map_err(SocksError::ServerError)?;
        return Err(SocksError::ServerError(SocksServerError::UnknownCommand(
            cmd.as_u8(),
        )));
    }

    let handle = match dial(target).await {
        Ok(handle) => handle,
        Err(err) => {
            proto
                .reply_error(&err.to_reply_error())
                .await
                .map_err(SocksError::ServerError)?;
            return Err(SocksError::ServerError(SocksServerError::Refused(
                "dial refused in-stream",
            )));
        }
    };

    let inner = proto
        .reply_success(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 0))
        .await
        .map_err(SocksError::ServerError)?;

    pump_bidi(inner, handle.read, handle.write).await;
    Ok(())
}

/// The pump handler over the minimal CONNECT path (the smoke test's
/// producer stand-in, the POC #1 shape). Awaited inline (R-02).
fn make_minimal_connect_handler(dial: DialFn) -> alkcall::channels::operations::OpenHandler {
    Arc::new(
        move |_input: serde_json::Value,
              _plan: Option<alkcall::channels::operations::ChannelPlan>,
              conn: CoreConnection,
              _auth: AuthContext| {
            let dial = Arc::clone(&dial);
            tokio::spawn(async move {
                let Some(bidi) = conn.accept_bi().await.ok() else {
                    tracing::error!("socks5 pump: accept_bi failed");
                    return;
                };
                if let Err(err) = serve_connect_minimal(bidi, dial).await {
                    tracing::warn!("socks5 session ended: {err:#}");
                }
            })
        },
    )
}

/// The producer-side registration mode (what [`wire_with`] registers
/// on the producer's serving registry post-hoc, W2's shape; the
/// single place the producer-side wiring lives, so the per-command
/// tasks only write tests).
#[derive(Clone)]
pub enum RegisterMode {
    /// The minimal CONNECT stand-in (the POC #1 inline shape) over
    /// 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>,
    },
    /// The spec'd assembly shape (producer.md §Wiring): the
    /// registration rides [`alksocks::producer::Socks5Producer::
    /// builder`] + [`alksocks::producer::Socks5Producer::register`]
    /// (the builder's `dial`/`egress`/`accept` setters and the
    /// delegated registration: the surface D-2 guards).
    ProducerBuilder {
        dial: DialFn,
        egress: Option<EgressFactory>,
        accept: Option<BindAcceptFactory>,
        bind_policy: Option<bool>,
    },
    /// No socks5 openable; only the generic channel ops (the
    /// unknown-op / registry-gate probe's shape).
    NoSocks5,
}

/// Register the SOCKS5 open op on a `ChannelCore` + the session's
/// dispatch registry: the alkcall 0.8.0 establishment surface
/// (`register_openable_with_establisher`) consumed for this crate's
/// shape. The producer task's `register_socks5_openable` succeeds
/// this helper (the smoke test's stand-in handler is replaced then;
/// the registration shape (spec, establisher, handler, auth,
/// timeout) stays).
pub fn register_socks5_openable_minimal(
    core: &ChannelCore,
    on_registry: &Arc<OperationRegistry>,
    auth: AuthContext,
    dial: DialFn,
    witness: IdentityWitness,
) -> Result<(), String> {
    core.register_openable_with_establisher(
        socks5_open_spec(),
        Some(socks5_session_establisher(witness)),
        make_minimal_connect_handler(dial),
        on_registry,
        auth,
        None,
    )
}

// --- The topology -----------------------------------------------------------

pub struct Topology {
    /// The producer's client (connect side, serving). Its manager
    /// allocated the socks5 channels (`ChannelSide::Connect`).
    pub producer: Arc<ChannelClient>,
    /// The consumer's manager (accept side): the initiator adopts
    /// the producer-allocated IDs here; probes reap via it.
    pub consumer_manager: ChannelManager,
    /// The consumer's channel-0 call connection: its surface for
    /// calling the producer's open op (and channel/close).
    pub consumer_call: Arc<CallConnection>,
    /// The producer's serving registry (the socks5 openable +
    /// generic channel ops live here; probes).
    pub producer_registry: Arc<OperationRegistry>,
    /// The consumer's serving registry (empty by default; the
    /// consumer serves no ops in the default topology).
    pub consumer_registry: Arc<OperationRegistry>,
    /// What the establisher's per-call auth carried (the CF-006
    /// probe; `None` until an open runs).
    pub identity_witness: IdentityWitness,
}

/// Wire the producer↔consumer topology over one duplex transport.
///
/// Identity wiring (the load-bearing POC #1 requirement): the
/// accept-side install hook calls `channel0_conn.set_identity`
/// before building the `CallConnection`; the serving dispatch reads
/// it as the caller identity (CF-005 (b)'s accept-side analogue;
/// without it, scope-gated opens fail closed with `FORBIDDEN`). The
/// transport identity attaches to the dialing connection before
/// `from_connection_with_serving` (the mTLS/QUIC posture) and
/// propagates automatically on the connect side.
///
/// `serving_identity` overrides the dispatch-resolved identity per
/// `ServingConfig.identity` (CF-005 (a) probe shape; `None` in the
/// primary path). `provider` resolves payload tokens (the
/// hub-forwarding path).
pub async fn wire_with(
    mode: RegisterMode,
    consumer_identity: Option<Identity>,
    serving_identity: Option<Identity>,
    provider: Arc<dyn IdentityProvider>,
) -> Topology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let consumer_serving_registry = Arc::new(OperationRegistry::new());
    let (consumer_call_tx, mut consumer_call_rx) =
        tokio::sync::mpsc::channel::<Arc<CallConnection>>(1);
    let (consumer_manager_tx, mut consumer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let identity_witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None));
    let fallback = Identity {
        id: "consumer-fallback".to_string(),
        scopes: vec![SOCKS5_OPEN_SCOPE.to_string()],
        resources: HashMap::new(),
    };
    let resolved = Arc::new(consumer_identity.unwrap_or(fallback));

    // --- consumer (accept side): ChannelsAdapter + capturing hook ---
    let resolved_for_hook = Arc::clone(&resolved);
    let serving_registry_for_hook = Arc::clone(&consumer_serving_registry);
    let install_hook: InstallChannelZero = Arc::new(
        move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| {
            let consumer_call_tx = consumer_call_tx.clone();
            let consumer_manager_tx = consumer_manager_tx.clone();
            let serving_registry = Arc::clone(&serving_registry_for_hook);
            let resolved = Arc::clone(&resolved_for_hook);
            tokio::spawn(async move {
                let channel0_bidi = match channel0_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(_) => return,
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                // The load-bearing POC #1 wiring requirement: attach
                // the resolved identity to channel 0 BEFORE building
                // the `CallConnection`; the serving dispatch reads
                // it as the caller identity; without it, scope-gated
                // opens fail closed (`FORBIDDEN "authentication
                // required"`). Explicit here; tested below.
                let _ = channel0_conn.set_identity((*resolved).clone());
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));

                // The consumer serves the ops pre-registered on its
                // serving registry (empty in the default topology;
                // the dispatcher keeps the call loop alive either
                // way; its read loop resolves the producer's
                // responses to the consumer's pending entries).
                let _ = consumer_call_tx.send(Arc::clone(&call_connection)).await;
                let _ = consumer_manager_tx.send(manager).await;
                let dp = Dispatcher::new(
                    serving_registry,
                    Arc::new(alkcall::core::auth::NoopIdentityProvider),
                );
                dp.serve_single_stream(call_connection, reader, writer)
                    .await;
            })
        },
    );

    // --- transport ------------------------------------------------------
    // The producer dials the transport (connect side): its connection
    // gets the transport identity BEFORE `from_connection_with_serving`
    // (the mTLS/QUIC posture; CF-005 (b), the primary caller-identity
    // path). The accept side resolves it out-of-band (modeled in the
    // hook above).
    let (producer_end, consumer_end) = tokio::io::duplex(64 * 1024);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);

    let identity_for_transport = (*resolved).clone();
    producer_conn
        .set_identity(identity_for_transport)
        .expect("transport identity set once");

    // The consumer (accept side) handles its end with the adapter +
    // install hook.
    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let consumer_conn_for_adapter = consumer_conn;
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            consumer_conn_for_adapter,
            &producer_transport_auth,
        )
        .await;
    });

    // --- producer (connect side, serving) -------------------------------
    let producer_client = ChannelClient::from_connection_with_serving(
        producer_conn,
        Some(ServingConfig {
            registry: Arc::clone(&producer_op_registry),
            identity_provider: provider,
            identity: serving_identity,
        }),
    )
    .await
    .expect("producer channel client init");
    let producer_client = Arc::new(producer_client);

    // The generic channel ops ride the producer's serving registry
    // (channel/close tears a channel down out-of-band per review 007's
    // recipe).
    let ops = ChannelOperations::with_default_policy(producer_client.manager().clone());
    ops.register_on(&producer_op_registry)
        .expect("register channel ops on producer");

    // Post-hoc openable registration on the producer's registry (W2):
    // the alkcall 0.8.0 establishment surface, consumed for this
    // crate's shape.
    if let RegisterMode::MinimalConnect(dial) = &mode {
        let core = ChannelCore::new(
            producer_client.manager().clone(),
            alkcall::channels::policy::default_policy(),
        );
        register_socks5_openable_minimal(
            &core,
            &producer_op_registry,
            AuthContext::anonymous(b"alk/socks5"),
            Arc::clone(dial),
            Arc::clone(&identity_witness),
        )
        .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)");
    }
    if let RegisterMode::ProducerBuilder {
        dial,
        egress,
        accept,
        bind_policy,
    } = &mode
    {
        let core = ChannelCore::new(
            producer_client.manager().clone(),
            alkcall::channels::policy::default_policy(),
        );
        let mut producer = alksocks::producer::Socks5Producer::builder().dial(Arc::clone(dial));
        if let Some(egress) = egress {
            producer = producer.egress(egress.clone());
        }
        if let Some(accept) = accept {
            producer = producer.accept(accept.clone());
        }
        if let Some(can_bind) = bind_policy {
            producer = producer.bind_policy(*can_bind);
        }
        producer
            .build()
            .register(
                &core,
                &producer_op_registry,
                AuthContext::anonymous(b"alk/socks5"),
                Some(Arc::clone(&identity_witness)),
            )
            .expect("register socks5 openable (the spec'd builder shape)");
    }

    // --- consumer pieces (captured from the hook) -----------------------
    let consumer_call = consumer_call_rx.recv().await.expect("consumer call conn");
    let consumer_manager = consumer_manager_rx.recv().await.expect("consumer manager");

    Topology {
        producer: producer_client,
        consumer_manager,
        consumer_call,
        producer_registry: Arc::clone(&producer_op_registry),
        consumer_registry: consumer_serving_registry,
        identity_witness,
    }
}

/// The default topology: the consumer's transport identity is the
/// 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::Producer {
            dial,
            egress: None,
            accept: None,
        },
        Some(consumer_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

// --- The consumer-side (forward) topology -------------------------------------

/// The consumer-session topology (`Socks5ChannelSession::connect` is
/// normative for this shape): the consumer dials the transport
/// (connect side; `ChannelClient::from_connection`, a pure consumer
/// serving nothing) and the producer accepts (the adapter + an
/// install hook registering the SOCKS5 openable over the injected
/// seams on the producer's serving registry).
pub struct ConsumerSideTopology {
    /// The consumer's own client (connect side, pure consumer). Its
    /// manager adopted the producer-allocated channel IDs; the
    /// session reaps via it.
    pub consumer: Arc<ChannelClient>,
    /// The producer's manager (accept side, serving): for asserting
    /// the handler-exit reaping (no leaked channel entries).
    pub producer_manager: ChannelManager,
    /// What the establisher's per-call auth carried (the CF-006
    /// probe; `None` until an open runs).
    pub identity_witness: IdentityWitness,
}

/// Wire the consumer-side topology over one duplex transport:
/// consumer dials (identity attached before `from_connection`: the
/// CF-005 (b) transport-identity path), producer accepts (the hook
/// attaches the resolved identity to channel 0 before the
/// `CallConnection`: the load-bearing POC #1 wiring requirement).
pub async fn wire_consumer_side_with(
    dial: DialFn,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    identity: Option<Identity>,
) -> ConsumerSideTopology {
    wire_consumer_side_with_provider(
        dial,
        egress,
        accept,
        identity,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

/// The consumer-side topology with the producer-side dispatch's
/// identity provider configured (the CF-005 precedence (1) seam: the
/// payload `auth_token` → `identity_provider` resolution path the
/// cred-bearing door's token sessions ride). Everything else matches
/// [`wire_consumer_side_with`].
pub async fn wire_consumer_side_with_provider(
    dial: DialFn,
    egress: Option<EgressFactory>,
    accept: Option<BindAcceptFactory>,
    identity: Option<Identity>,
    provider: Arc<dyn IdentityProvider>,
) -> ConsumerSideTopology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let identity_witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None));
    let witness_for_hook = Arc::clone(&identity_witness);
    let resolved = Arc::new(identity.unwrap_or_else(consumer_identity));

    let resolved_for_hook = Arc::clone(&resolved);
    let provider_for_hook = Arc::clone(&provider);
    let dial_for_hook = Arc::clone(&dial);
    let install_hook: InstallChannelZero = Arc::new(
        move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| {
            let registry = Arc::clone(&producer_op_registry);
            let manager_tx = producer_manager_tx.clone();
            let witness = Arc::clone(&witness_for_hook);
            let provider = Arc::clone(&provider_for_hook);
            let resolved = Arc::clone(&resolved_for_hook);
            let dial = Arc::clone(&dial_for_hook);
            let egress = egress.clone();
            let accept = accept.clone();
            tokio::spawn(async move {
                let channel0_bidi = match channel0_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(_) => return,
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let _ = channel0_conn.set_identity((*resolved).clone());
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                let core =
                    ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy());
                let ops = ChannelOperations::with_default_policy(manager.clone());
                ops.register_on(&registry)
                    .expect("register channel ops on producer");
                alksocks::producer::register_socks5_openable(
                    &core,
                    &registry,
                    AuthContext::anonymous(b"alk/socks5"),
                    Some(dial),
                    egress,
                    accept,
                    Some(witness),
                )
                .expect("register socks5 openable (the consumer-side topology)");
                let _ = manager_tx.send(manager).await;
                let dp = Dispatcher::new(registry, provider);
                dp.serve_single_stream(call_connection, reader, writer)
                    .await;
            })
        },
    );

    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    consumer_conn
        .set_identity((*resolved).clone())
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    let consumer_client = ChannelClient::from_connection(consumer_conn)
        .await
        .expect("consumer channel client init");
    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");

    ConsumerSideTopology {
        consumer: Arc::new(consumer_client),
        producer_manager,
        identity_witness,
    }
}

/// The default consumer-side topology: the CONNECT path over the
/// given dial, the consumer identity on the transport.
pub async fn wire_consumer_side(dial: DialFn) -> ConsumerSideTopology {
    wire_consumer_side_with(dial, None, None, Some(consumer_identity())).await
}

// --- The composed suite's topology (the alktunnels direct op) -------

/// The tunnel-direct topology (the composed suite's shape): the
/// consumer dials the transport (a full `ChannelClient`, the
/// composed egress's direct-op caller) and the producer accepts (the
/// adapter + an install hook registering the alktunnels direct op,
/// `register_tunnel_direct_openable` over the given dial, on the
/// producer's serving registry). The resolved identity carries
/// `tunnel:direct` (the direct op's scope gate).
pub struct TunnelDirectTopology {
    /// The consumer's client: `TunnelSession::open_direct` rides it.
    pub consumer: Arc<ChannelClient>,
    /// The producer's manager (accept side, serving): probes.
    pub producer_manager: ChannelManager,
}

/// Wire the tunnel-direct topology over one duplex transport (the
/// consumer-side topology's shape, the direct op in place of the
/// SOCKS5 openable; the identity carries `tunnel:direct`).
pub async fn wire_tunnel_direct(dial: alktunnels::producer::DialFn) -> TunnelDirectTopology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let resolved = Identity {
        id: "composed-egress-consumer".to_string(),
        scopes: vec![alktunnels::params::TUNNEL_DIRECT_SCOPE.to_string()],
        resources: HashMap::new(),
    };
    let resolved = Arc::new(resolved);
    let resolved_for_hook = Arc::clone(&resolved);
    let dial_for_hook = Arc::clone(&dial);
    let install_hook: InstallChannelZero = Arc::new(
        move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| {
            let registry = Arc::clone(&producer_op_registry);
            let manager_tx = producer_manager_tx.clone();
            let resolved = Arc::clone(&resolved_for_hook);
            let dial = Arc::clone(&dial_for_hook);
            tokio::spawn(async move {
                let channel0_bidi = match channel0_conn.accept_bi().await {
                    Ok(s) => s,
                    Err(_) => return,
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let _ = channel0_conn.set_identity((*resolved).clone());
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                let core =
                    ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy());
                alktunnels::producer::register_tunnel_direct_openable(
                    &core,
                    &registry,
                    AuthContext::anonymous(b"alk/tunnel"),
                    dial,
                    None,
                )
                .expect("register tunnel direct openable (the composed topology)");
                let _ = manager_tx.send(manager).await;
                let dp = Dispatcher::new(
                    registry,
                    Arc::new(alkcall::core::auth::NoopIdentityProvider),
                );
                dp.serve_single_stream(call_connection, reader, writer)
                    .await;
            })
        },
    );

    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    consumer_conn
        .set_identity((*resolved).clone())
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    let consumer_client = ChannelClient::from_connection(consumer_conn)
        .await
        .expect("consumer channel client init");
    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");

    TunnelDirectTopology {
        consumer: Arc::new(consumer_client),
        producer_manager,
    }
}

// --- The raw RFC 1928 client helper -----------------------------------------

/// A hand-rolled minimal RFC 1928 client over any
/// `AsyncRead + AsyncWrite` (greeting → method selection → command
/// with target → reply read). The differential suite's established
/// pattern: the raw helper drives the producer without depending on
/// the consumer session types (which land later).
pub struct RawRfcClient<S> {
    pub stream: S,
}

impl<S> RawRfcClient<S>
where
    S: AsyncRead + AsyncWrite + Unpin,
{
    /// Wrap a stream (a channel's `BiStream`, a duplex half, anything
    /// substrate-agnostic).
    pub fn new(stream: S) -> Self {
        Self { stream }
    }

    /// Greeting (`05 01 00`: version 5, one method, no-auth) → the
    /// method-selection read (`05 00`). Returns the selected method
    /// byte.
    pub async fn greet(&mut self) -> Result<u8, std::io::Error> {
        raw_greet(&mut self.stream).await
    }

    /// Write a command request for `target` (the request form:
    /// `VER CMD RSV ATYP DST.ADDR DST.PORT`).
    pub async fn request(
        &mut self,
        cmd: Socks5Command,
        target: &TargetAddr,
    ) -> Result<(), std::io::Error> {
        raw_request(&mut self.stream, &cmd, target).await
    }

    /// Read one reply (the 4-byte head + the BND.ADDR per its ATYP).
    /// Returns `(reply_code, bnd_addr_bytes)`: the address bytes as
    /// the RFC form carries them (ATYP + addr + port).
    pub async fn read_reply(&mut self) -> Result<(u8, Vec<u8>), std::io::Error> {
        let (code, _atyp, addr) = raw_read_reply(&mut self.stream).await?;
        Ok((code, addr))
    }

    /// The full CONNECT round-trip: greet, request, reply. Returns
    /// the reply code.
    pub async fn connect(&mut self, target: &TargetAddr) -> Result<u8, std::io::Error> {
        let method = self.greet().await?;
        assert_eq!(
            method,
            consts::SOCKS5_AUTH_METHOD_NONE,
            "harness uses no-auth"
        );
        self.request(Socks5Command::TCPConnect, target).await?;
        Ok(self.read_reply().await?.0)
    }
}

// --- Tests ------------------------------------------------------------------

// --- The composed accept's topology (the bind-first tunnel producer) ---

/// The bind-first topology (the composed accept suite's shape,
/// alktunnels' `wire_listener_bindfirst` adapted): the opener dials
/// the transport (the SOCKS5 consumer side; its `ChannelClient` is
/// what `TunnelListener::open` and the SOCKS5 BIND client ride) and
/// the tunnel producer accepts (the adapter + an install hook
/// registering the alktunnels bind-first openable over the given
/// bind/accept-loop injections AND the SOCKS5 producer openable with
/// the composed accept factory). One connection carries both
/// protocols' ops; the producer serves `alk/tunnel`'s open op and the
/// forwarded op, plus the SOCKS5 open op, on one registry.
///
/// The opener identity carries `tunnel:open` + `tunnel:forwarded` +
/// `socks5:open` + `socks5:bind` (the four gates the composed BIND
/// story crosses: the listen open, the forwarded ingress, the SOCKS5
/// session open, the BIND command gate).
pub struct BindFirstTopology {
    /// The opener's client: `TunnelListener::open` + the SOCKS5 BIND
    /// session + the composed accept factory all ride it.
    pub opener: Arc<ChannelClient>,
    /// The opener's serving registry: the forwarded op lands here
    /// (via `register_forwarded_serving`).
    pub opener_registry: Arc<OperationRegistry>,
    /// The producer's manager (accept side, serving): reaping probes.
    pub producer_manager: ChannelManager,
    /// The producer's serving registry (probes).
    pub producer_registry: Arc<OperationRegistry>,
    /// What the bind-first establisher's per-call auth carried (the
    /// CF-006 probe).
    pub identity_witness: Arc<tokio::sync::Mutex<Option<String>>>,
}

/// Wire the bind-first topology over one duplex transport. The
/// producer's install hook registers, on its serving registry, the
/// alktunnels bind-first openable (over the given `bind` + the given
/// accept loop) and the SOCKS5 openable over a composed accept factory
/// built from `register_forwarded_serving` on the OPENER side... but
/// the forwarded op serves on the opener's registry, which only
/// exists after the opener's client is built. The hook therefore
/// defers the forwarded registration: the test wires it post-hoc via
/// [`alksocks::composed::register_forwarded_serving`] on
/// `opener_registry` (W2's post-hoc posture: the dispatcher reads
/// through the shared `Arc<OperationRegistry>` per dispatch).
pub async fn wire_bindfirst_composed(
    registry: alktunnels::producer::ResourceRegistry,
    bind: alktunnels::producer::BindFn,
    accept_loop: alktunnels::producer::BindFirstAcceptLoop,
    accept_factory: Option<BindAcceptFactory>,
    opener_identity: Identity,
) -> BindFirstTopology {
    let producer_op_registry = Arc::new(OperationRegistry::new());
    let opener_op_registry = Arc::new(OperationRegistry::new());
    let registry_for_hook = Arc::clone(&producer_op_registry);
    let opener_registry_for_struct = Arc::clone(&opener_op_registry);
    let (producer_manager_tx, mut producer_manager_rx) =
        tokio::sync::mpsc::channel::<ChannelManager>(1);
    let identity_witness: Arc<tokio::sync::Mutex<Option<String>>> =
        Arc::new(tokio::sync::Mutex::new(None));
    let bind_witness = Arc::clone(&identity_witness);
    let opener_identity_for_hook = opener_identity.clone();

    // --- producer (accept side, serving): adapter + install hook ---
    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
        let registry_arc = Arc::clone(&registry_for_hook);
        let resource_registry = registry.clone();
        let bind = Arc::clone(&bind);
        let accept_loop = Arc::clone(&accept_loop);
        let accept = accept_factory.clone();
        let witness = Arc::clone(&bind_witness);
        let manager_tx = producer_manager_tx.clone();
        let caller = opener_identity_for_hook.clone();
        tokio::spawn(async move {
            let channel0_bidi = match channel0_conn.accept_bi().await {
                Ok(s) => s,
                Err(_) => return,
            };
            let (writer, reader) = split_single_stream(channel0_bidi);
            // The accept-side identity posture (the load-bearing POC #1
            // requirement): attach the resolved identity to channel 0
            // BEFORE building the `CallConnection`.
            let _ = channel0_conn.set_identity(caller);
            let call_connection = Arc::new(CallConnection::new_single_stream(
                channel0_conn,
                Arc::clone(&writer),
            ));

            let core =
                ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy());
            let ops = ChannelOperations::with_default_policy(manager.clone());
            ops.register_on(&registry_arc)
                .expect("register channel ops on producer");

            // The forwarded-open caller (the call plane, alktunnels
            // harness's shape): call + adopt on the serving session's
            // `CallConnection`.
            let call_for_opener = Arc::clone(&call_connection);
            let manager_for_opener = manager.clone();
            let open_forwarded: alktunnels::producer::ForwardedOpener =
                Arc::new(move |forwarded: alktunnels::TunnelForwardedParams| {
                    let call = Arc::clone(&call_for_opener);
                    let manager = manager_for_opener.clone();
                    Box::pin(async move {
                        let input = serde_json::to_value(&forwarded).map_err(|e| {
                            alktunnels::TunnelEstablishError::HandlerError(format!(
                                "forwarded params serialize: {e}"
                            ))
                        })?;
                        let response = call.call(alktunnels::OP_TUNNEL_FORWARDED, input).await;
                        let out = response.result.map_err(|e| {
                            alktunnels::TunnelEstablishError::HandlerError(format!(
                                "forwarded open: {}",
                                e.message
                            ))
                        })?;
                        let channel_id =
                            out.get("channel_id")
                                .and_then(|v| v.as_u64())
                                .ok_or_else(|| {
                                    alktunnels::TunnelEstablishError::HandlerError(
                                        "forwarded open reply missing channel_id".to_string(),
                                    )
                                })? as u32;
                        let (send, recv) = manager
                            .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
                            .await
                            .map_err(|e| {
                                alktunnels::TunnelEstablishError::HandlerError(format!(
                                    "forwarded adopt: {e}"
                                ))
                            })?;
                        Ok((send, recv))
                    })
                });

            alktunnels::producer::register_tunnel_bindfirst_openable(
                &core,
                &resource_registry,
                &registry_arc,
                AuthContext::anonymous(b"alk/tunnel"),
                alktunnels::producer::BindFirstInjections {
                    bind,
                    accept_loop,
                    open_forwarded,
                },
                Some(witness),
            )
            .expect("register tunnel bindfirst openable");

            // The SOCKS5 openable rides the same registry when the
            // accept factory is available at hook time (the
            // static-factory shape). When the factory closes over
            // topology pieces that only exist after the opener's
            // client is built (the composed accept's factory needs the
            // opener's `ChannelClient`), the registration defers to
            // the post-hoc W2 shape; the caller registers it on the
            // returned `producer_registry` (the dispatcher reads
            // through the shared `Arc<OperationRegistry>` per
            // dispatch).
            if let Some(accept) = accept {
                let producer = alksocks::producer::Socks5Producer::builder()
                    .accept(accept)
                    .build();
                producer
                    .register(
                        &core,
                        &registry_arc,
                        AuthContext::anonymous(b"alk/socks5"),
                        None,
                    )
                    .expect("register socks5 openable (the composed accept topology)");
            }

            let _ = manager_tx.send(manager).await;
            let dp = Dispatcher::new(
                registry_arc,
                Arc::new(alkcall::core::auth::NoopIdentityProvider),
            );
            dp.serve_single_stream(call_connection, reader, writer)
                .await;
        })
    });

    // --- transport ------------------------------------------------------
    let (opener_end, producer_end) = tokio::io::duplex(64 * 1024);
    let opener_conn = CoreConnection::from_bidi(opener_end, b"alk/channels".to_vec(), None);
    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
    opener_conn
        .set_identity(opener_identity.clone())
        .expect("transport identity set once");

    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
    let _adapter_task = tokio::spawn(async move {
        let _ = alkcall::core::types::ProtocolHandler::handle(
            &adapter,
            producer_conn,
            &producer_transport_auth,
        )
        .await;
    });

    // --- opener (connect side): client + serving (the forwarded op
    // registers post-hoc, per the type docs) -----------------------------
    let opener_client = ChannelClient::from_connection_with_serving(
        opener_conn,
        Some(ServingConfig {
            registry: Arc::clone(&opener_op_registry),
            identity_provider: Arc::new(alkcall::core::auth::NoopIdentityProvider),
            identity: None,
        }),
    )
    .await
    .expect("opener channel client init");

    let producer_manager = producer_manager_rx
        .recv()
        .await
        .expect("producer manager captured");

    BindFirstTopology {
        opener: Arc::new(opener_client),
        opener_registry: opener_registry_for_struct,
        producer_manager,
        producer_registry: producer_op_registry,
        identity_witness,
    }
}

/// The composed suite's opener identity: the four gates the composed
/// BIND story crosses (`tunnel:open` + `tunnel:forwarded` + `socks5:open`
/// + `socks5:bind`).
pub fn composed_opener_identity() -> Identity {
    Identity {
        id: "composed-opener".to_string(),
        scopes: vec![
            alktunnels::TUNNEL_OPEN_SCOPE.to_string(),
            alktunnels::TUNNEL_FORWARDED_SCOPE.to_string(),
            SOCKS5_OPEN_SCOPE.to_string(),
            SOCKS5_BIND_SCOPE.to_string(),
        ],
        resources: HashMap::new(),
    }
}

/// [`composed_opener_identity`] minus the forwarded grant: the
/// per-accept isolation probe's identity (the forwarded open fails
/// FORBIDDEN; the accept is consumed, the listener lives).
pub fn composed_scopeless_opener_identity() -> Identity {
    Identity {
        id: "composed-opener-scopeless".to_string(),
        scopes: vec![
            alktunnels::TUNNEL_OPEN_SCOPE.to_string(),
            SOCKS5_OPEN_SCOPE.to_string(),
            SOCKS5_BIND_SCOPE.to_string(),
        ],
        resources: HashMap::new(),
    }
}

/// The composed suite's full identity: the five gates the composed
/// three-command story crosses (`tunnel:open` + `tunnel:direct` +
/// `tunnel:forwarded` + `socks5:open` + `socks5:bind`); the assembly
/// suite's producer registration serves the direct op (CONNECT +
/// ASSOCIATE's tunnels) beside the bind-first op (BIND's listener).
pub fn composed_full_opener_identity() -> Identity {
    Identity {
        id: "composed-opener-full".to_string(),
        scopes: vec![
            alktunnels::TUNNEL_OPEN_SCOPE.to_string(),
            alktunnels::TUNNEL_DIRECT_SCOPE.to_string(),
            alktunnels::TUNNEL_FORWARDED_SCOPE.to_string(),
            SOCKS5_OPEN_SCOPE.to_string(),
            SOCKS5_BIND_SCOPE.to_string(),
        ],
        resources: HashMap::new(),
    }
}

// --- The composed accept's shared machinery (review 002 N-5) ----------------
//
// The bind/accept-queue stand-ins and the channel-set probe the
// composed suites consume, single-sourced here now that two suites
// (composed-accept + assembly-integration) drive the same shapes.

/// The resource the bind-first producer serves (the composed suites'
/// only subscription, the registry namespace gate).
pub const COMPOSED_RESOURCE: &str = "bind";

/// The bound address the test bind reports (the "OS-chosen" stand-in;
/// the test's queue listener owns it; fidelity is the assert's job).
pub const COMPOSED_BOUND_PORT: u16 = 42113;

/// The peer address the accepted pairs report (the "app server"; the
/// forwarded op's `peer` param carries it to reply#2).
pub const COMPOSED_PEER_PORT: u16 = 55231;

/// The test bind (the assembly's bind seam): reports the test's
/// bound address + an opaque listener witness.
pub fn composed_test_bind() -> alktunnels::producer::BindFn {
    Arc::new(|_substrate, _backing| {
        Box::pin(async {
            Ok(alktunnels::producer::BoundListener {
                listener: Arc::new(()) as Arc<dyn std::any::Any + Send + Sync>,
                bound: alktunnels::params::SubstrateAddr::HostPort {
                    host: "127.0.0.1".to_string(),
                    port: COMPOSED_BOUND_PORT,
                },
            })
        })
    })
}

/// The accept-queue pair the test accept loop pops from (the alktunnels
/// listener test's shape: a VecDeque + Notify, one accepted pair per
/// dispatcher call).
pub type ComposedAcceptQueue = (
    Arc<
        tokio::sync::Mutex<
            std::collections::VecDeque<(
                alktunnels::params::SubstrateAddr,
                alktunnels::producer::TargetHandle,
            )>,
        >,
    >,
    Arc<tokio::sync::Notify>,
);

pub fn composed_fresh_queue() -> ComposedAcceptQueue {
    (
        Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new())),
        Arc::new(tokio::sync::Notify::new()),
    )
}

/// An accepted "connection" for the test accept loop: a duplex pair
/// with an echo task on the far end (the app server's socket stand-in;
/// the data plane's echo confirms the full round trip).
fn composed_accepted_halves(
    peer_addr: (&str, u16),
) -> (
    alktunnels::params::SubstrateAddr,
    alktunnels::producer::TargetHandle,
) {
    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
    let (mut t_read, mut t_write) = tokio::io::split(target_side);
    tokio::spawn(async move {
        let mut buf = vec![0u8; 8192];
        loop {
            match t_read.read(&mut buf).await {
                Ok(0) | Err(_) => return,
                Ok(n) => {
                    if t_write.write_all(&buf[..n]).await.is_err() {
                        return;
                    }
                }
            }
        }
    });
    let (c_read, c_write) = tokio::io::split(consumer_side);
    (
        alktunnels::params::SubstrateAddr::HostPort {
            host: peer_addr.0.to_string(),
            port: peer_addr.1,
        },
        alktunnels::producer::TargetHandle {
            read: Box::new(c_read),
            write: Box::new(c_write),
        },
    )
}

/// The test accept loop (the dispatcher invokes the closure per
/// accept): pop one accepted pair, or park until notified.
pub fn composed_queue_loop(
    queue: ComposedAcceptQueue,
) -> alktunnels::producer::BindFirstAcceptLoop {
    let (queue, notify) = queue;
    Arc::new(move |_listener| {
        let queue = Arc::clone(&queue);
        let notify = Arc::clone(&notify);
        Box::pin(async move {
            let popped = {
                let notified = std::pin::pin!(notify.notified());
                let mut q = queue.lock().await;
                if let Some(pair) = q.pop_front() {
                    Some(pair)
                } else {
                    drop(q);
                    notified.await;
                    let mut q = queue.lock().await;
                    q.pop_front()
                }
            };
            popped.ok_or(alktunnels::TunnelEstablishError::DialFailed(
                "listener closed".to_string(),
            ))
        })
    })
}

/// Push an accepted pair into the test's accept queue.
pub async fn composed_push_accept(queue: &ComposedAcceptQueue, peer: (&str, u16)) {
    let (peer_addr, handle) = composed_accepted_halves(peer);
    queue.0.lock().await.push_back((peer_addr, handle));
    queue.1.notify_waiters();
}

/// The composed suites' no-leak probe: the opener side ends with
/// channel 0 (+ the live channels the test names); the producer side
/// ends with channel 0, the SOCKS5 channels the test names, AND any
/// odd id: the producer-side ADOPTED forwarded channels
/// (opener-allocated ids, adopted inside the producer's
/// `open_forwarded` caller; per-session ownership per alktunnels
/// ADR-005; their entries reap at transport EOF, not at the
/// listener's teardown; upstream's own bind-first suites make the
/// same concession in `no_stray_channels`).
pub async fn assert_composed_channels(
    topo: &BindFirstTopology,
    opener_extra: &[u32],
    producer_extra: &[u32],
) {
    let opener_ids = topo.opener.manager().channel_ids();
    let producer_ids = topo.producer_manager.channel_ids();
    let opener_ok = opener_ids
        .iter()
        .all(|id| *id == 0 || opener_extra.contains(id));
    let producer_ok = producer_ids
        .iter()
        .all(|id| *id == 0 || *id % 2 == 1 || producer_extra.contains(id));
    assert!(
        opener_ok && producer_ok,
        "channel sets: opener={opener_ids:?} producer={producer_ids:?}"
    );
}

#[cfg(test)]
mod tests {
    use super::*;
    use alksocks::egress::{BindAcceptFactoryError, EgressFactory};

    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();
        assert_eq!(open.scopes, vec![SOCKS5_OPEN_SCOPE.to_string()]);
        assert_eq!(open.id, "consumer");

        let bind = bind_identity();
        assert_eq!(bind.scopes, vec![SOCKS5_BIND_SCOPE.to_string()]);
        assert!(!bind.scopes.contains(&SOCKS5_OPEN_SCOPE.to_string()));

        let both = both_scopes_identity();
        assert!(both.scopes.contains(&SOCKS5_OPEN_SCOPE.to_string()));
        assert!(both.scopes.contains(&SOCKS5_BIND_SCOPE.to_string()));
        assert_eq!(both.scopes.len(), 2, "both grants, explicitly");
    }

    #[test]
    fn test_id_provider_resolves_the_scoped_token() {
        let provider = TestIdProvider;
        let token = AuthToken {
            raw: TEST_AUTH_TOKEN.as_bytes().to_vec(),
        };
        assert_eq!(
            provider.resolve_from_token(&token).map(|i| i.id),
            Some("consumer".to_string())
        );
        assert_eq!(
            provider
                .resolve_from_token(&token)
                .and_then(|i| i.scopes.into_iter().next()),
            Some(SOCKS5_OPEN_SCOPE.to_string())
        );
        assert!(provider
            .resolve_from_token(&AuthToken {
                raw: b"other".to_vec(),
            })
            .is_none());
        assert!(provider.resolve_from_fingerprint("any").is_none());
    }

    #[tokio::test]
    async fn echo_dial_round_trips_through_pump() {
        let handle = echo_dial()(TargetAddr::Domain("example.com".into(), 443))
            .await
            .expect("dial resolves");
        let (mut probe_far, probe_near) = tokio::io::duplex(64 * 1024);
        // The pump feeds the handle against the probe's near end,
        // exactly the data-plane shape the producer handler runs.
        let pump = tokio::spawn(pump_bidi(probe_near, handle.read, handle.write));
        probe_far.write_all(b"ping").await.expect("write");
        let mut buf = [0u8; 4];
        probe_far.read_exact(&mut buf).await.expect("echoed back");
        assert_eq!(&buf, b"ping");
        drop(probe_far);
        let (c2p, p2c) = pump.await.expect("pump completes");
        assert_eq!((c2p, p2c), (4, 4), "both pumps moved the probe bytes");
    }

    #[tokio::test]
    async fn refusing_dial_maps_to_its_reply_code() {
        let dial = refusing_dial(SocksDialError::NotAllowed("policy".into()));
        let err = dial(TargetAddr::Domain("example.com".into(), 443))
            .await
            .expect_err("dial refuses");
        assert_eq!(
            err.to_reply_error().as_u8(),
            ReplyError::ConnectionNotAllowed.as_u8()
        );
    }

    #[tokio::test]
    async fn echo_egress_echoes_inner_bytes_and_parks_when_idle() {
        let mut egress = echo_egress()().await.expect("egress builds");
        let payload = Bytes::from_static(b"query");
        egress
            .send(
                TargetAddr::Ip("10.0.0.1:53".parse().unwrap()),
                payload.clone(),
            )
            .await
            .expect("send succeeds");
        // The echo comes back as inner bytes (addr encoding + data).
        let reply = egress
            .next_reply()
            .await
            .expect("next_reply")
            .expect("a reply is ready");
        let expected = {
            let mut v = TargetAddr::Ip("10.0.0.1:53".parse().unwrap())
                .to_be_bytes()
                .expect("addr encodes");
            v.extend_from_slice(&payload);
            Bytes::from(v)
        };
        assert_eq!(reply, expected);

        // Readiness contract: with nothing in flight, `next_reply`
        // parks (`Pending`), never a hot `Ok(None)`.
        let mut poll = egress.next_reply();
        tokio::time::timeout(std::time::Duration::from_millis(50), &mut poll)
            .await
            .expect_err("next_reply stays Pending when nothing is ready");
    }

    #[tokio::test]
    async fn echo_accept_resolves_with_halves_and_echoes() {
        let accept = echo_accept()(TargetAddr::Domain("example.com".into(), 2121))
            .await
            .expect("factory resolves");
        assert_eq!(
            accept.bind_addr(),
            SocketAddr::from(([127, 0, 0, 1], ECHO_BIND_PORT))
        );
        let mut handle = accept.accept().await.expect("accept resolves");
        assert_eq!(
            handle.peer,
            TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 7], 44444)))
        );
        // Echo round-trip through the accepted halves.
        handle.write.write_all(b"ping").await.expect("write");
        let mut buf = [0u8; 4];
        handle.read.read_exact(&mut buf).await.expect("echoed");
        assert_eq!(&buf, b"ping");
    }

    #[tokio::test]
    async fn failing_accept_variants_refuse_at_their_stages() {
        // Factory failure → pre-reply#1.
        let factory =
            refusing_accept_factory(BindAcceptFactoryError::Refused("no listener".into()));
        let err = factory(TargetAddr::Domain("example.com".into(), 2121))
            .await
            .expect_err("factory refuses");
        assert_eq!(
            err.to_reply_error().as_u8(),
            ReplyError::ConnectionRefused.as_u8()
        );

        // Accept-stage failure → post-reply#1 (reply#2).
        let factory = refusing_accept(BindAcceptError::NotAllowed("policy".into()));
        let accept = factory(TargetAddr::Domain("example.com".into(), 2121))
            .await
            .expect("factory resolves; the accept future refuses");
        let err = accept.accept().await.expect_err("accept refuses");
        assert_eq!(
            err.to_reply_error().as_u8(),
            ReplyError::ConnectionNotAllowed.as_u8()
        );
    }

    #[tokio::test]
    async fn refusing_egress_refuses_at_associate_setup() {
        let factory = refusing_egress(EgressError::Unreachable("no egress".into()));
        let err = match factory().await {
            Err(err) => err,
            Ok(_) => panic!("the factory must refuse"),
        };
        assert_eq!(
            err.to_reply_error().as_u8(),
            ReplyError::NetworkUnreachable.as_u8()
        );
    }

    #[test]
    fn seam_closures_have_the_poc_compile_shapes() {
        assert_send_sync::<DialFn>();
        assert_send_sync::<EgressFactory>();
        assert_send_sync::<BindAcceptFactory>();
        assert_send::<BindAccept>();
    }

    // -- the raw RFC client helper ---------------------------------------

    /// A hand-rolled minimal RFC server over a duplex: greeting →
    /// method selection (`05 00`) → reply read (the differential
    /// suite's mock-server pattern); it is the raw client's counterparty.
    async fn hand_rolled_server(mut stream: tokio::io::DuplexStream, reply_code: u8) {
        let mut greeting = [0u8; 3];
        stream.read_exact(&mut greeting).await.expect("greeting");
        assert_eq!(greeting, [0x05, 0x01, 0x00]);
        stream
            .write_all(&[consts::SOCKS5_VERSION, consts::SOCKS5_AUTH_METHOD_NONE])
            .await
            .expect("method selection");
        // Read the command request head (VER CMD RSV ATYP) + the rest
        // per ATYP (a domain here, the helper's test target form).
        let mut head = [0u8; 4];
        stream.read_exact(&mut head).await.expect("command head");
        assert_eq!(head[1], consts::SOCKS5_CMD_TCP_CONNECT);
        match head[3] {
            consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => {
                let mut len = [0u8; 1];
                stream.read_exact(&mut len).await.expect("domain len");
                let mut rest = vec![0u8; len[0] as usize + 2];
                stream.read_exact(&mut rest).await.expect("domain + port");
            }
            consts::SOCKS5_ADDR_TYPE_IPV4 => {
                let mut rest = [0u8; 6];
                stream.read_exact(&mut rest).await.expect("v4 + port");
            }
            other => panic!("test wrote an unexpected address type {other}"),
        }
        // Reply with a static v4 BND.ADDR.
        let addr = SocketAddr::from(([127, 0, 0, 1], 1080));
        stream
            .write_all(&new_reply(&ReplyError::from_u8(reply_code), addr))
            .await
            .expect("reply written");
    }

    /// The raw client round-trips greeting + command + reply against
    /// a hand-rolled server over a duplex (the task's acceptance
    /// criterion: the raw helper is producer-independent).
    #[tokio::test]
    async fn raw_rfc_client_round_trips_against_a_hand_rolled_server() {
        let (client_end, server_end) = tokio::io::duplex(1024);
        let _server = tokio::spawn(hand_rolled_server(
            server_end,
            consts::SOCKS5_REPLY_SUCCEEDED,
        ));
        let mut raw = RawRfcClient::new(client_end);

        let code = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            raw.connect(&TargetAddr::Domain("echo.test".into(), 443)),
        )
        .await
        .expect("round-trip timed out")
        .expect("the raw client completes");
        assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);

        // The refusal path: the server's reply code surfaces verbatim.
        let (client_end, server_end) = tokio::io::duplex(1024);
        let server = tokio::spawn(hand_rolled_server(
            server_end,
            consts::SOCKS5_REPLY_CONNECTION_REFUSED,
        ));
        let mut raw = RawRfcClient::new(client_end);
        let code = raw
            .connect(&TargetAddr::Ip("127.0.0.1:1".parse().unwrap()))
            .await
            .expect("round-trip completes");
        assert_eq!(code, consts::SOCKS5_REPLY_CONNECTION_REFUSED);
        let _ = server.await;
    }

    // -- the smoke test ---------------------------------------------------

    /// The full loop: harness up → producer registered → consumer
    /// opens a channel → raw RFC CONNECT conversation → echo
    /// round-trip. The producer path is the minimal CONNECT stand-in
    /// (the POC #1 inline shape); the producer task replaces it.
    #[tokio::test]
    async fn smoke_connect_echo_round_trip_over_the_harness() {
        let topo = wire(echo_dial()).await;

        // The consumer calls the producer's open op on channel 0 and
        // adopts the producer-allocated channel id (the alkcall 0.8.0
        // establishment surface, consumed end-to-end).
        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;
        assert!(channel_id >= 1);
        let (send, recv) = topo
            .consumer_manager
            .adopt_channel(channel_id, SOCKS5_ALPN, None)
            .await
            .expect("adopt the producer-allocated id");

        // The raw RFC client drives the producer through the channel:
        // pump the RFC conversation between the raw client's duplex
        // and the channel halves.
        let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024);
        tokio::spawn(pump_bidi(
            stream_self,
            Box::new(recv) as Box<dyn AsyncRead + Send + Unpin>,
            Box::new(send) as Box<dyn AsyncWrite + Send + Unpin>,
        ));

        let mut raw = RawRfcClient::new(stream_peer);
        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);

        // Echo round-trip through the data plane: whatever the client
        // sends is echoed by the dialed in-process target.
        raw.stream
            .write_all(b"ping")
            .await
            .expect("write the probe");
        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 producer-side channel is alive; the identity the
        // establisher witnessed is the consumer's (CF-006).
        assert_eq!(
            topo.identity_witness.lock().await.as_deref(),
            Some("consumer")
        );
    }

    /// The scope-less identity fails closed: an identity with no
    /// `socks5:open` gets the registry's `FORBIDDEN` (the open gate
    /// is the boundary; ADR-006); this is the POC #1 wiring requirement
    /// tested explicitly (the hook's `set_identity` is what made the
    /// allowed path work above; omitting it yields the same
    /// `FORBIDDEN`, fail-closed).
    #[tokio::test]
    async fn scopeless_identity_fails_closed_with_forbidden() {
        // A bare identity with no scopes at all.
        let bare = Identity {
            id: "scopeless".to_string(),
            scopes: vec![],
            resources: std::collections::HashMap::new(),
        };
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: None,
                accept: None,
            },
            Some(bare),
            None,
            Arc::new(alkcall::core::auth::NoopIdentityProvider),
        )
        .await;

        let response = topo
            .consumer_call
            .call(OP_SOCKS5_OPEN, Socks5Params::as_json())
            .await;
        let err = response
            .result
            .expect_err("a scope-less identity fails closed");
        assert_eq!(err.code, "FORBIDDEN");
        // Fail-closed is fail-closed: no channel anywhere.
        assert!(topo
            .consumer_manager
            .channel_ids()
            .into_iter()
            .all(|id| id == 0));
        assert!(topo
            .producer
            .manager()
            .channel_ids()
            .into_iter()
            .all(|id| id == 0));
    }

    /// The establisher's per-call auth witness: the opener identity
    /// id arrives on the establisher's `AuthContext` (CF-006); the
    /// seam the producer task maps SOCKS5 credentials/ACL onto.
    #[tokio::test]
    async fn establisher_witnesses_the_opener_identity() {
        let topo = wire(echo_dial()).await;
        let response = topo
            .consumer_call
            .call(OP_SOCKS5_OPEN, Socks5Params::as_json())
            .await;
        assert!(response.result.is_ok(), "the channel opens");
        tokio::time::timeout(std::time::Duration::from_secs(5), async {
            loop {
                let seen = topo.identity_witness.lock().await.clone();
                if seen.is_some() {
                    assert_eq!(seen.as_deref(), Some("consumer"));
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
        })
        .await
        .expect("the establisher ran and witnessed the opener");
    }

    /// The harness topology sanity: the producer's manager allocated
    /// the channel id (the serving side allocates; the initiator
    /// adopts; ADR-047 §5).
    #[tokio::test]
    async fn producer_allocates_consumer_adopts() {
        let topo = wire(echo_dial()).await;
        let response = topo
            .consumer_call
            .call(OP_SOCKS5_OPEN, Socks5Params::as_json())
            .await;
        let out = response.result.expect("open");
        let channel_id = out.get("channel_id").and_then(|v| v.as_u64()).expect("id") as u32;
        assert_eq!(channel_id % 2, 1, "the connect side allocates odd ids");
        assert!(!topo.consumer_manager.has_channel(channel_id));
        assert!(topo.producer.manager().has_channel(channel_id));
        let (_send, _recv) = topo
            .consumer_manager
            .adopt_channel(channel_id, SOCKS5_ALPN, None)
            .await
            .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);
    }

    /// ASSOCIATE refuses RFC-faithfully when the egress seam is not
    /// wired (`CommandNotSupported` 0x07, ADR-008's missing-callback
    /// class) while CONNECT still serves the same registration (the
    /// missing-callback posture, over channels).
    #[tokio::test]
    async fn producer_associate_refuses_0x07_connect_still_serves() {
        let topo = wire(echo_dial()).await;

        // ASSOCIATE refuses (no egress seam wired).
        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);

        // 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");
    }

    // -- the producer ASSOCIATE datagram stage over channels -------------
    // (the frame helpers ride the common module: `udp_frame` is
    // `datagram_frame`, `udp_unframe` is `decode_datagram`.)

    /// The ASSOCIATE conversation over a channel: open + adopt +
    /// pump, greet, request, reply, asserting the sentinel
    /// `127.0.0.1:0` BND.ADDR (ADR-003's RFC-compatibility surface).
    async fn raw_associate(
        topo: &Topology,
    ) -> (
        RawRfcClient<tokio::io::DuplexStream>,
        tokio::task::JoinHandle<(u64, u64)>,
    ) {
        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_SUCCEEDED, "the associate reply");
        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"
        );
        (raw, pump)
    }

    /// The datagram stage end to end over channels (the POC #2
    /// shape): sentinel reply → datagrams round-trip through the echo
    /// egress in order, one association; EOF (channel shutdown) ends
    /// the association and a fresh associate succeeds immediately
    /// (the POC's test 4); the empty-payload datagram is legal (never
    /// EOF); a garbage-ATYP frame ends the association loud.
    #[tokio::test]
    async fn producer_udp_associate_datagram_stage_over_channels() {
        init_trace();
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: Some(echo_egress()),
                accept: None,
            },
            Some(consumer_identity()),
            None,
            Arc::new(alkcall::core::auth::NoopIdentityProvider),
        )
        .await;

        // Associate: the sentinel reply, then the datagram stage.
        let (mut raw, pump) = raw_associate(&topo).await;
        let target = TargetAddr::Ip("10.0.0.1:53".parse().unwrap());

        // Multiple datagrams, in order, one association.
        for i in 0..5u8 {
            raw.stream
                .write_all(&udp_frame(&target, &[i; 10]))
                .await
                .expect("datagram sent");
        }
        for i in 0..5u8 {
            let mut buf = vec![0u8; 2 + 7 + 10];
            tokio::time::timeout(
                std::time::Duration::from_secs(10),
                raw.stream.read_exact(&mut buf),
            )
            .await
            .expect("echo timed out")
            .expect("reply frame read");
            let (src, payload) = udp_unframe(&buf);
            assert_eq!(src, target, "the echo reports the destination");
            assert_eq!(&payload[..], &vec![i; 10][..], "in-order echo");
        }

        // The empty-payload datagram: legal, never EOF (the F-2
        // invariant at payload level).
        raw.stream
            .write_all(&udp_frame(&target, b""))
            .await
            .expect("empty datagram sent");
        let mut buf = vec![0u8; 2 + 7];
        tokio::time::timeout(
            std::time::Duration::from_secs(10),
            raw.stream.read_exact(&mut buf),
        )
        .await
        .expect("empty echo timed out")
        .expect("reply frame read");
        let (_, payload) = udp_unframe(&buf);
        assert!(payload.is_empty(), "the empty datagram round-trips");

        // A garbage-ATYP frame ends the association loud: EOF on the
        // next read.
        raw.stream
            .write_all(&[0x00, 0x04, 0x7f, 1, 2, 3])
            .await
            .expect("garbage frame sent");
        let mut buf = [0u8; 16];
        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!("EOF expected after an undecodable frame; got {n} bytes"),
        }
        drop(raw);
        tokio::time::timeout(std::time::Duration::from_secs(10), pump)
            .await
            .expect("the stage pump ends")
            .expect("join");
    }

    /// Channel EOF ends the association and the connection stays
    /// healthy (the POC's test 4, over channels): after one
    /// association's channel is torn down, a fresh associate succeeds
    /// immediately and round-trips.
    #[tokio::test]
    async fn producer_udp_associate_eof_teardown_connection_stays_healthy() {
        init_trace();
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: Some(echo_egress()),
                accept: None,
            },
            Some(consumer_identity()),
            None,
            Arc::new(alkcall::core::auth::NoopIdentityProvider),
        )
        .await;

        // First association: EOF ends it (half-close the channel).
        let (raw, pump) = raw_associate(&topo).await;
        drop(raw);
        tokio::time::timeout(std::time::Duration::from_secs(10), pump)
            .await
            .expect("the stage pump ends on channel EOF")
            .expect("join");

        // The connection is healthy: a fresh associate round-trips.
        let (mut raw, _pump) = raw_associate(&topo).await;
        let target = TargetAddr::Ip("10.0.0.2:53".parse().unwrap());
        raw.stream
            .write_all(&udp_frame(&target, b"fresh"))
            .await
            .expect("datagram sent");
        let mut buf = vec![0u8; 2 + 7 + 5];
        tokio::time::timeout(
            std::time::Duration::from_secs(10),
            raw.stream.read_exact(&mut buf),
        )
        .await
        .expect("fresh echo timed out")
        .expect("reply frame read");
        let (_, payload) = udp_unframe(&buf);
        assert_eq!(&payload[..], b"fresh");
    }

    /// The egress-factory refusal maps to reply#1's mapped code over
    /// channels (the pre-reply refusal path); no phantom datagram
    /// plane.
    #[tokio::test]
    async fn producer_udp_associate_factory_refusal_maps_to_reply_code() {
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: Some(refusing_egress(EgressError::Unreachable(
                    "no egress".into(),
                ))),
                accept: None,
            },
            Some(consumer_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::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_NETWORK_UNREACHABLE,
            "the factory refusal maps to reply#1's code"
        );
    }

    /// CONNECT + ASSOCIATE on the same registration over channels
    /// (the regression guard, extended: one registration, both
    /// commands).
    #[tokio::test]
    async fn producer_connect_and_associate_on_the_same_registration() {
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: Some(echo_egress()),
                accept: None,
            },
            Some(consumer_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"tcp").await.expect("write");
        let mut buf = [0u8; 3];
        raw.stream.read_exact(&mut buf).await.expect("echo");
        assert_eq!(&buf, b"tcp");
        drop(raw);

        // ASSOCIATE serves the same registration.
        let (mut raw, _pump) = raw_associate(&topo).await;
        let target = TargetAddr::Ip("10.0.0.3:53".parse().unwrap());
        raw.stream
            .write_all(&udp_frame(&target, b"udp"))
            .await
            .expect("datagram sent");
        let mut buf = vec![0u8; 2 + 7 + 3];
        raw.stream.read_exact(&mut buf).await.expect("echo");
        let (_, payload) = udp_unframe(&buf);
        assert_eq!(&payload[..], b"udp");
    }

    /// 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, with 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`, so 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 (POC #5 test 7): one
    /// registration, all three commands; CONNECT serves, BIND serves
    /// under its gate, ASSOCIATE serves with the echo egress. Fresh
    /// channel per session.
    #[tokio::test]
    async fn producer_three_command_regression_guard() {
        let topo = wire_with(
            RegisterMode::Producer {
                dial: echo_dial(),
                egress: Some(echo_egress()),
                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 serves: the sentinel reply + one datagram
        // round-trip.
        let (mut raw, _pump) = raw_associate(&topo).await;
        let target = TargetAddr::Ip("10.0.0.9:53".parse().unwrap());
        raw.stream
            .write_all(&udp_frame(&target, b"all3"))
            .await
            .expect("datagram sent");
        let mut buf = vec![0u8; 2 + 7 + 4];
        tokio::time::timeout(
            std::time::Duration::from_secs(10),
            raw.stream.read_exact(&mut buf),
        )
        .await
        .expect("echo timed out")
        .expect("reply frame read");
        let (_, payload) = udp_unframe(&buf);
        assert_eq!(&payload[..], b"all3");
    }

    /// 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.
    #[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);
        }
    }
}