alksocks/tests/harness.rs

//! Test harness — the producer↔consumer wiring over a
//! `tokio::io::duplex` carrying the channels 8-byte chunk format.
//!
//! 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` — the
//!   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/`) — it stays out of the wasm build.

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, Egress, EgressError,
    EgressFactory, SocksDialError, TargetHandle,
};
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;

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

/// 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 ---------------------------------------------

/// An echo dial closure (the ADR-004 injection point under test): a
/// duplex per dial — the "target" echoes bytes until EOF. The
/// producer-side pump gets split `DuplexStream` halves; a spawned
/// task echoes the other end. No sockets, no `local` feature.
pub fn echo_dial() -> DialFn {
    Arc::new(move |_target: TargetAddr| {
        Box::pin(async {
            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(TargetHandle {
                read: Box::new(read_half),
                write: Box::new(write_half),
            })
        })
    })
}

/// A dial that always refuses with the given [`SocksDialError`]
/// variant (the refusal-mapping tests' dial — the refusal rides
/// in-stream as the mapped RFC reply code). The variant is
/// reconstructed per call (the seam errors are not `Clone`).
pub fn refusing_dial(err: SocksDialError) -> DialFn {
    Arc::new(move |_target: TargetAddr| {
        let err = match &err {
            SocksDialError::Refused(m) => SocksDialError::Refused(m.clone()),
            SocksDialError::NetworkUnreachable(m) => SocksDialError::NetworkUnreachable(m.clone()),
            SocksDialError::HostUnreachable(m) => SocksDialError::HostUnreachable(m.clone()),
            SocksDialError::NotAllowed(m) => SocksDialError::NotAllowed(m.clone()),
            SocksDialError::General(m) => SocksDialError::General(m.clone()),
        };
        Box::pin(async move { Err(err) })
    })
}

/// An echo egress: datagrams sent to it echo back (the POC #2
/// shape). The readiness contract is honored: `next_reply` parks on
/// the queue's `recv()` — `Pending` when nothing is ready — never an
/// instantly-ready `Ok(None)`.
pub fn echo_egress() -> EgressFactory {
    alksocks::egress::egress_factory(|| {
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Bytes>();
        Ok(Box::new(ChanEgress {
            tx,
            rx: Mutex::new(rx),
        }) as Box<dyn Egress + Send>)
    })
}

use tokio::sync::Mutex;

/// The in-process egress behind [`echo_egress`]: `send` queues the
/// inner bytes for echo; `next_reply` parks on the queue (`Pending`
/// until a datagram arrives — the readiness contract, harness-modeled).
struct ChanEgress {
    tx: tokio::sync::mpsc::UnboundedSender<Bytes>,
    rx: Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
}

impl Egress for ChanEgress {
    fn send(
        &mut self,
        addr: TargetAddr,
        payload: Bytes,
    ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
        Box::pin(async move {
            // Echo: the reply's inner bytes are `addr + data` —
            // exactly what `next_reply` returns (the wire layer's
            // length prefix is applied exactly once, by the frame
            // writer, never here).
            let mut inner = addr
                .to_be_bytes()
                .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
            inner.extend_from_slice(&payload);
            self.tx
                .send(Bytes::from(inner))
                .map_err(|_| EgressError::General("egress closed".into()))
        })
    }

    fn next_reply(&mut self) -> futures::future::BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
        Box::pin(async move {
            // Real async park: `Pending` until a datagram arrives (the
            // readiness contract — never an instantly-ready `Ok(None)`,
            // which would starve a biased `select!` into livelock).
            match self.rx.lock().await.recv().await {
                Some(inner) => Ok(Some(inner)),
                // All senders dropped: end the association (an `Err`,
                // never a hot `Ok(None)`).
                None => Err(EgressError::General("egress closed".into())),
            }
        })
    }
}

/// A failing egress factory (the ASSOCIATE-refusal probe): refuses
/// with the given error, which the producer maps to reply#1's code.
/// The variant is reconstructed per call (the seam errors are not
/// `Clone`).
pub fn refusing_egress(err: EgressError) -> EgressFactory {
    Arc::new(move || {
        let err = match &err {
            EgressError::Refused(m) => EgressError::Refused(m.clone()),
            EgressError::Unreachable(m) => EgressError::Unreachable(m.clone()),
            EgressError::General(m) => EgressError::General(m.clone()),
        };
        Box::pin(async move { Err(err) })
    })
}

/// 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()` — no plan 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::Bug(
                "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),
    /// 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 — 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)");
    }

    // --- 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 minimal
/// CONNECT stand-in over the given dial.
pub async fn wire(dial: DialFn) -> Topology {
    wire_with(
        RegisterMode::MinimalConnect(dial),
        Some(consumer_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await
}

// --- 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 —
    /// 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> {
        self.stream.write_all(&[0x05, 0x01, 0x00]).await?;
        let mut chosen = [0u8; 2];
        self.stream.read_exact(&mut chosen).await?;
        Ok(chosen[1])
    }

    /// 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> {
        let addr = target
            .to_be_bytes()
            .map_err(|e| std::io::Error::other(format!("target encode: {e}")))?;
        let mut req = vec![consts::SOCKS5_VERSION, cmd.as_u8(), 0x00];
        req.extend_from_slice(&addr);
        self.stream.write_all(&req).await?;
        self.stream.flush().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 mut head = [0u8; 4];
        self.stream.read_exact(&mut head).await?;
        let addr_bytes: Vec<u8> = match head[3] {
            consts::SOCKS5_ADDR_TYPE_IPV4 => {
                let mut a = [0u8; 6];
                self.stream.read_exact(&mut a).await?;
                a.to_vec()
            }
            consts::SOCKS5_ADDR_TYPE_IPV6 => {
                let mut a = [0u8; 18];
                self.stream.read_exact(&mut a).await?;
                a.to_vec()
            }
            consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME => {
                let mut len = [0u8; 1];
                self.stream.read_exact(&mut len).await?;
                let mut a = vec![0u8; len[0] as usize + 2];
                self.stream.read_exact(&mut a).await?;
                let mut full = vec![len[0]];
                full.extend_from_slice(&a);
                full
            }
            other => {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("unknown address type {other}"),
                ))
            }
        };
        Ok((head[1], addr_bytes))
    }

    /// 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 ------------------------------------------------------------------

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

    fn assert_send_sync<T: Send + Sync>() {}
    fn assert_send<T: Send>() {}

    #[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) — 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) — 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 — no scopes at all.
        let bare = Identity {
            id: "scopeless".to_string(),
            scopes: vec![],
            resources: std::collections::HashMap::new(),
        };
        let topo = wire_with(
            RegisterMode::MinimalConnect(echo_dial()),
            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));
    }
}