alksocks 0.3.2

//! The cross-cutting integration suite: the whole protocol core's
//! behavior pinned in one place (the batch-1 closing task; the
//! architecture-sync check's test half).
//!
//! The per-command tasks own their verticals (tests/consumer-*.rs and
//! the `src/` unit suites); this suite owns the cross-cutting
//! invariants:
//!
//! - **One registration, all three commands** (the POC #2/#5
//!   regression guard, complete): CONNECT + ASSOCIATE + BIND all
//!   served on the same `channels/socks5/sub` op, interleaved live
//!   sessions on one channels connection.
//! - **Concurrency**: mixed concurrent sessions over one channels
//!   connection (the mux/demux under load; the POCs tested
//!   sequentially), pinned in both allocation directions (the raw
//!   topology where the producer's serving side allocates the channel
//!   ids, and the consumer-side topology where the consumer does).
//! - **The teardown matrix** (wire.md §Sentinels and lifecycle, the
//!   normative source): a client half-close ends a CONNECT data plane
//!   cleanly; ASSOCIATE EOF ends the association; a BIND client-gone
//!   teardown releases the accept; a fresh session succeeds
//!   immediately after each teardown (no connection poisoning).
//! - **The identity/scope matrix** (ADR-006, end to end):
//!   `socks5:open`-only (CONNECT + ASSOCIATE serve, BIND refused 0x02
//!   at the command gate), `socks5:bind`-only (fail-closed at the
//!   open gate, all three commands; ADR-006's open gate requires
//!   `socks5:open` unconditionally and the two grants are never
//!   implied), both scopes (everything, granted explicitly).
//! - **The readiness contract** re-pinned under concurrent load (the
//!   producer-associate task owns the unit-level test): a hot
//!   `Ok(None)` egress across concurrent associations must not starve
//!   the datagram stage's channel branch.

mod harness;

use std::future::Future;
use std::net::SocketAddr;
use std::sync::Arc;

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

use alksocks::consumer::{AssociateSession, BindSession, Socks5ChannelSession};
use alksocks::egress::{Egress, EgressError, EgressFactory};
use alksocks::error::SocksSessionError;
use alksocks::params::Socks5Params;
use alksocks::rfc1928::consts;
use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
use alksocks::rfc1928::Socks5Command;

use harness::{
    bind_identity, both_scopes_identity, consumer_identity, echo_accept, echo_dial, echo_egress,
    wire_consumer_side_with, wire_with, ConsumerSideTopology, RawRfcClient, RegisterMode, Topology,
    ECHO_BIND_PORT,
};

const TIMEOUT_SECS: u64 = 10;

type RawPump = tokio::task::JoinHandle<(u64, u64)>;
type RawClient = RawRfcClient<tokio::io::DuplexStream>;
type RoundTrip<'a> = std::pin::Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>>;

async fn within<T>(fut: impl Future<Output = T>) -> T {
    tokio::time::timeout(std::time::Duration::from_secs(TIMEOUT_SECS), fut)
        .await
        .expect("operation timed out")
}

// --- Topology builders ------------------------------------------------------

/// The full raw topology: all three seams wired, the both-scopes
/// identity (everything serves).
async fn full_raw_topology() -> Topology {
    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
}

// --- Raw-session helpers (the harness's raw-RFC pattern) ---------------------

/// Open a socks5 channel + adopt it + pump the raw RFC client through
/// it (the smoke test's opening shape).
async fn open_adopt_pump(topo: &Topology) -> (RawClient, RawPump) {
    let response = topo
        .consumer_call
        .call(alksocks::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, alksocks::SOCKS5_ALPN, None)
        .await
        .expect("adopt the producer-allocated id");
    let (stream_peer, stream_self) = tokio::io::duplex(64 * 1024);
    let pump = tokio::spawn(alkcall::channels::pump::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 ASSOCIATE conversation over a fresh channel: greet → request →
/// the sentinel `127.0.0.1:0` reply (ADR-003).
async fn raw_associate(topo: &Topology) -> (RawClient, RawPump) {
    let (mut raw, pump) = open_adopt_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 BIND conversation over a fresh channel through the echo
/// accept: greet → request → reply#1 (the advertised port) → reply#2
/// (the accepted peer, the harness's fixed stand-in).
async fn raw_bind(topo: &Topology, host: &str) -> (RawClient, RawPump) {
    let (mut raw, pump) = open_adopt_pump(topo).await;
    raw.greet().await.expect("greet");
    raw.request(
        Socks5Command::TCPBind,
        &TargetAddr::Domain(host.into(), ECHO_BIND_PORT),
    )
    .await
    .expect("request written");
    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"
    );
    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");
    (raw, pump)
}

async fn connect_round_trip(topo: &Topology, host: String, tag: String) -> Result<(), String> {
    let (mut raw, _pump) = open_adopt_pump(topo).await;
    let code = raw
        .connect(&TargetAddr::Domain(host, 443))
        .await
        .map_err(|e| e.to_string())?;
    if code != consts::SOCKS5_REPLY_SUCCEEDED {
        return Err(format!("reply code {code}"));
    }
    raw.stream
        .write_all(tag.as_bytes())
        .await
        .map_err(|e| e.to_string())?;
    let mut buf = vec![0u8; tag.len()];
    raw.stream
        .read_exact(&mut buf)
        .await
        .map_err(|e| e.to_string())?;
    if buf != tag.as_bytes() {
        return Err("echo mismatch".into());
    }
    Ok(())
}

/// A raw ASSOCIATE round-trip: sentinel reply + one datagram echoed
/// in order.
async fn associate_round_trip(
    topo: &Topology,
    octet: u8,
    tag: &'static [u8],
) -> Result<(), String> {
    let (mut raw, _pump) = raw_associate(topo).await;
    let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, octet], 53)));
    raw.stream
        .write_all(&udp_frame(&target, tag))
        .await
        .map_err(|e| e.to_string())?;
    let mut buf = vec![0u8; 2 + 7 + tag.len()];
    raw.stream
        .read_exact(&mut buf)
        .await
        .map_err(|e| e.to_string())?;
    let (src, payload) = udp_unframe(&buf);
    if src != target {
        return Err("echo source mismatch".into());
    }
    if &payload[..] != tag {
        return Err("echo payload mismatch".into());
    }
    Ok(())
}

/// A raw BIND round-trip: reply#1 + reply#2 + one pass-through echo
/// exchange on the data plane.
async fn bind_round_trip(
    topo: &Topology,
    host: impl ToString,
    tag: &'static [u8],
) -> Result<(), String> {
    let (mut raw, _pump) = raw_bind(topo, &host.to_string()).await;
    raw.stream.write_all(tag).await.map_err(|e| e.to_string())?;
    let mut buf = vec![0u8; tag.len()];
    raw.stream
        .read_exact(&mut buf)
        .await
        .map_err(|e| e.to_string())?;
    if buf != tag {
        return Err("echo mismatch".into());
    }
    Ok(())
}

/// A raw CONNECT round-trip boxed for mixed concurrent joining.
fn tcp_fut<'a>(topo: &'a Topology, host: String, tag: String) -> RoundTrip<'a> {
    Box::pin(connect_round_trip(topo, host, tag))
}

/// A raw ASSOCIATE round-trip boxed for mixed concurrent joining.
fn udp_fut<'a>(topo: &'a Topology, octet: u8, tag: &'static [u8]) -> RoundTrip<'a> {
    Box::pin(associate_round_trip(topo, octet, tag))
}

/// A raw BIND round-trip boxed for mixed concurrent joining.
fn bind_fut<'a>(topo: &'a Topology, host: String, tag: &'static [u8]) -> RoundTrip<'a> {
    Box::pin(bind_round_trip(topo, host, tag))
}

// --- The ASSOCIATE wire-frame helpers (the common module's shapes) -----------

use harness::{parked_accept_factory as parked_accept, udp_frame, udp_unframe};

// --- Seam stand-ins ----------------------------------------------------------

// The parked accept factory rides the common module (imported as
// `parked_accept` above, the deterministic quiet-wait probe; the
// advertised bind address is the common module's 2121).

/// The hot egress (the readiness contract's violator, the
/// producer-associate unit test's shape): `next_reply` drains one
/// queued reply if any, then returns an instantly-ready `Ok(None)`
/// instead of parking. The stage must tolerate this under load: a
/// biased select starved into livelock would hang every session here.
struct HotNoneEgress {
    echo: tokio::sync::mpsc::UnboundedSender<bytes::Bytes>,
    replies: tokio::sync::Mutex<tokio::sync::mpsc::UnboundedReceiver<bytes::Bytes>>,
}

impl Egress for HotNoneEgress {
    fn send(
        &mut self,
        addr: TargetAddr,
        payload: bytes::Bytes,
    ) -> futures::future::BoxFuture<'_, Result<(), EgressError>> {
        Box::pin(async move {
            let mut inner = addr
                .to_be_bytes()
                .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
            inner.extend_from_slice(&payload);
            self.echo
                .send(bytes::Bytes::from(inner))
                .map_err(|_| EgressError::General("egress closed".into()))
        })
    }

    fn next_reply(
        &mut self,
    ) -> futures::future::BoxFuture<'_, Result<Option<bytes::Bytes>, EgressError>> {
        Box::pin(async move {
            match self.replies.lock().await.try_recv() {
                Ok(inner) => Ok(Some(inner)),
                Err(_) => Ok(None),
            }
        })
    }
}

fn hot_none_egress() -> EgressFactory {
    Arc::new(|| {
        Box::pin(async {
            let (echo, replies) = tokio::sync::mpsc::unbounded_channel();
            Ok(Box::new(HotNoneEgress {
                echo,
                replies: tokio::sync::Mutex::new(replies),
            }) as Box<dyn Egress + Send>)
        })
    })
}

// --- The consumer-session probes ----------------------------------------------

/// The no-leaked-channels probe (the per-command suites' shape): both
/// managers hold only channel 0; the producer's entry reaps by its
/// open-op wrapper; the probe polls past the reap race.
async fn no_leaked_channels(topo: &ConsumerSideTopology) {
    let deadline = std::time::Duration::from_secs(5);
    let consumer_ids = topo.consumer.manager().channel_ids();
    assert!(
        consumer_ids.iter().all(|&id| id == 0),
        "consumer leaked channel entries: {consumer_ids:?}"
    );
    tokio::time::timeout(deadline, async {
        loop {
            let producer_ids = topo.producer_manager.channel_ids();
            if producer_ids.iter().all(|&id| id == 0) {
                return;
            }
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
    })
    .await
    .expect("the producer's handler-exit teardown reaps the entry");
}

/// A `FORBIDDEN`-carrying open failure (the fail-closed probe's
/// expectation shape).
fn expect_forbidden(err: &SocksSessionError) -> bool {
    match err {
        SocksSessionError::Open(open) => match open.open_ref() {
            alkcall::channels::client::ChannelOpenError::CallFailed { error } => {
                error.code == "FORBIDDEN"
            }
            _ => false,
        },
        _ => false,
    }
}

/// A consumer CONNECT round-trip, result-shaped for concurrent
/// joining.
async fn consumer_connect_round_trip(
    topo: &ConsumerSideTopology,
    host: String,
    tag: String,
) -> Result<(), String> {
    let mut session = Socks5ChannelSession::connect(
        topo.consumer.as_ref(),
        (host.as_str(), 443u16).to_target_addr().unwrap(),
    )
    .await
    .map_err(|e| format!("connect: {e}"))?;
    session
        .split()
        .expect("pipe")
        .1
        .write_all(tag.as_bytes())
        .await
        .map_err(|e| e.to_string())?;
    let (mut read, _) = session.split().expect("pipe");
    let mut buf = vec![0u8; tag.len()];
    read.read_exact(&mut buf).await.map_err(|e| e.to_string())?;
    if buf != tag.as_bytes() {
        return Err("echo mismatch".into());
    }
    session.close().await;
    Ok(())
}

/// A consumer ASSOCIATE round-trip: one datagram echoed in order on
/// the framed stage.
async fn consumer_associate_round_trip(
    topo: &ConsumerSideTopology,
    octet: u8,
    tag: &'static [u8],
) -> Result<(), String> {
    let mut assoc = AssociateSession::associate(topo.consumer.as_ref())
        .await
        .map_err(|e| format!("associate: {e}"))?;
    let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 1, octet], 53)));
    assoc
        .send_datagram(&target, tag)
        .await
        .map_err(|e| format!("send: {e}"))?;
    let (src, payload) = within(assoc.recv_datagram())
        .await
        .map_err(|e| format!("recv: {e}"))?
        .expect("a datagram (not eof)");
    if src != target {
        return Err("echo source mismatch".into());
    }
    if &payload[..] != tag {
        return Err("echo payload mismatch".into());
    }
    assoc.close().await;
    Ok(())
}

/// A consumer CONNECT round-trip boxed for mixed concurrent joining.
fn consumer_tcp_fut<'a>(
    topo: &'a ConsumerSideTopology,
    host: String,
    tag: String,
) -> RoundTrip<'a> {
    Box::pin(consumer_connect_round_trip(topo, host, tag))
}

/// A consumer ASSOCIATE round-trip boxed for mixed concurrent joining.
fn consumer_udp_fut<'a>(
    topo: &'a ConsumerSideTopology,
    octet: u8,
    tag: &'static [u8],
) -> RoundTrip<'a> {
    Box::pin(consumer_associate_round_trip(topo, octet, tag))
}

/// A consumer BIND round-trip boxed for mixed concurrent joining.
fn consumer_bind_fut<'a>(
    topo: &'a ConsumerSideTopology,
    host: String,
    tag: &'static [u8],
) -> RoundTrip<'a> {
    Box::pin(consumer_bind_round_trip(topo, host, tag))
}

/// A consumer BIND round-trip: reply#1 + reply#2 + one echo exchange
/// on the pass-through plane.
async fn consumer_bind_round_trip(
    topo: &ConsumerSideTopology,
    host: String,
    tag: &'static [u8],
) -> Result<(), String> {
    let mut session = BindSession::bind(
        topo.consumer.as_ref(),
        (host.as_str(), ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .map_err(|e| format!("bind: {e}"))?;
    let mut plane = session
        .await_inbound()
        .await
        .map_err(|e| format!("reply#2: {e}"))?
        .expect("the data plane");
    plane.write_all(tag).await.map_err(|e| e.to_string())?;
    plane.flush().await.map_err(|e| e.to_string())?;
    let mut buf = vec![0u8; tag.len()];
    within(plane.read_exact(&mut buf))
        .await
        .map_err(|e| e.to_string())?;
    if buf != tag {
        return Err("echo mismatch".into());
    }
    drop(plane);
    session.close().await;
    Ok(())
}

// --- The cross-cutting suite ---------------------------------------------------

/// One registration, all three commands, interleaved live sessions on
/// one channels connection (the POC #2/#5 regression guard,
/// complete): all three sessions are opened back to back and are live
/// simultaneously; each is exercised mid-life of the others; a
/// teardown ends only its own session (the others keep serving, the
/// mux/demux isolation).
#[tokio::test]
async fn one_registration_all_three_commands_interleaved() {
    let topo = full_raw_topology().await;

    // Three interleaved opens on the one op; all three sessions live
    // at once (CONNECT, ASSOCIATE, BIND: fresh channel per session,
    // convention 10).
    let (mut connect_raw, connect_pump) = open_adopt_pump(&topo).await;
    let (mut assoc_raw, assoc_pump) = raw_associate(&topo).await;
    let (mut bind_raw, bind_pump) = raw_bind(&topo, "bind.test").await;

    // The CONNECT session's RFC handshake (the other two helpers ran
    // theirs on their own channels).
    connect_raw.greet().await.expect("greet");
    connect_raw
        .request(
            Socks5Command::TCPConnect,
            &TargetAddr::Domain("interleaved.test".into(), 443),
        )
        .await
        .expect("request written");
    let (code, _addr) = connect_raw.read_reply().await.expect("reply read");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);

    // Exercise each mid-life of the others.
    connect_raw.stream.write_all(b"tcp1").await.expect("write");
    let mut buf = [0u8; 4];
    connect_raw.stream.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"tcp1");

    let udp_target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 1], 53)));
    assoc_raw
        .stream
        .write_all(&udp_frame(&udp_target, b"udp1"))
        .await
        .expect("datagram");
    let mut frame_buf = vec![0u8; 2 + 7 + 4];
    assoc_raw
        .stream
        .read_exact(&mut frame_buf)
        .await
        .expect("frame");
    let (_, payload) = udp_unframe(&frame_buf);
    assert_eq!(&payload[..], b"udp1");

    bind_raw.stream.write_all(b"bin1").await.expect("write");
    bind_raw.stream.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"bin1");

    // A teardown ends only its own session.
    drop(connect_raw);
    within(connect_pump).await.expect("the CONNECT pump ends");

    // ASSOCIATE + BIND still serve (no connection poisoning).
    assoc_raw
        .stream
        .write_all(&udp_frame(&udp_target, b"still"))
        .await
        .expect("datagram");
    let mut frame_buf = vec![0u8; 2 + 7 + 5];
    assoc_raw
        .stream
        .read_exact(&mut frame_buf)
        .await
        .expect("frame");
    let (_, payload) = udp_unframe(&frame_buf);
    assert_eq!(&payload[..], b"still");

    bind_raw.stream.write_all(b"alive").await.expect("write");
    let mut buf = [0u8; 5];
    bind_raw.stream.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"alive");

    drop(assoc_raw);
    drop(bind_raw);
    within(assoc_pump).await.expect("associate pump");
    within(bind_pump).await.expect("bind pump");
}

/// Concurrency over one channels connection (the mux/demux under
/// load): nine concurrent sessions of mixed command types, three
/// CONNECTs, three ASSOCIATEs, three BINDs, interleaved by the
/// executor, all riding the one registration. The producer's serving
/// side allocates the channel ids (the raw topology's direction).
#[tokio::test]
async fn concurrent_mixed_sessions_over_one_connection() {
    let topo = full_raw_topology().await;

    const UDP_TAGS: [&[u8]; 3] = [b"udp-zero0", b"udp-one!1", b"udp-two!2"];
    const BIND_TAGS: [&[u8]; 3] = [b"bind-z0", b"bind-o1", b"bind-t2"];
    let mut futs: Vec<RoundTrip> = Vec::new();
    for i in 0..3usize {
        futs.push(tcp_fut(&topo, format!("c{i}.test"), format!("tcp{i}")));
        futs.push(udp_fut(&topo, 10 + i as u8, UDP_TAGS[i]));
        futs.push(bind_fut(&topo, format!("b{i}.test"), BIND_TAGS[i]));
    }
    let results = futures::future::join_all(futs).await;
    for (i, result) in results.into_iter().enumerate() {
        result.unwrap_or_else(|e| panic!("concurrent session {i} failed: {e}"));
    }

    // The connection is healthy after the burst: one more of each.
    connect_round_trip(&topo, "after.test".into(), "again".into())
        .await
        .expect("fresh connect after the load");
    associate_round_trip(&topo, 20, b"again")
        .await
        .expect("fresh associate after the load");
    bind_round_trip(&topo, "after.test", b"again")
        .await
        .expect("fresh bind after the load");
}

/// Concurrency through the typed consumer sessions over one channels
/// connection: six concurrent sessions (two of each command) driven
/// by the real session types. The consumer dials here (the
/// consumer-side topology), so the consumer's side allocates the
/// channel ids (the mux/demux pinned in the other allocation
/// direction. No channel entry leaks on either side.
#[tokio::test]
async fn concurrent_consumer_sessions_over_one_connection() {
    let topo = wire_consumer_side_with(
        echo_dial(),
        Some(echo_egress()),
        Some(echo_accept()),
        Some(both_scopes_identity()),
    )
    .await;

    const UDP_TAGS: [&[u8]; 2] = [b"cu-zero00", b"cu-one!!1"];
    const BIND_TAGS: [&[u8]; 2] = [b"cb-zero", b"cb-one!"];
    let mut futs: Vec<RoundTrip> = Vec::new();
    for i in 0..2usize {
        futs.push(consumer_tcp_fut(
            &topo,
            format!("cc{i}.test"),
            format!("t{i}"),
        ));
        futs.push(consumer_udp_fut(&topo, 30 + i as u8, UDP_TAGS[i]));
        futs.push(consumer_bind_fut(
            &topo,
            format!("cb{i}.test"),
            BIND_TAGS[i],
        ));
    }
    let results = futures::future::join_all(futs).await;
    for (i, result) in results.into_iter().enumerate() {
        result.unwrap_or_else(|e| panic!("consumer session {i} failed: {e}"));
    }

    no_leaked_channels(&topo).await;
}

/// The teardown matrix, CONNECT row (wire.md §Sentinels and
/// lifecycle): a client half-close ends the data plane cleanly (the
/// producer's pump exits) and a fresh session of every command
/// succeeds immediately after (no connection poisoning).
#[tokio::test]
async fn teardown_connect_half_close_then_every_command_serves_fresh() {
    let topo = full_raw_topology().await;

    let (mut raw, pump) = open_adopt_pump(&topo).await;
    let code = raw
        .connect(&TargetAddr::Domain("drop.test".into(), 443))
        .await
        .expect("connect");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
    raw.stream.write_all(b"pre").await.expect("write");
    let mut buf = [0u8; 3];
    raw.stream.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"pre");

    // The client half-close: the producer's pump exits (the data
    // plane is over; the connection stays healthy).
    drop(raw);
    within(pump)
        .await
        .expect("the pump ends on the client half-close");

    // Fresh sessions of every command succeed immediately.
    let (mut raw, _pump) = open_adopt_pump(&topo).await;
    let code = raw
        .connect(&TargetAddr::Domain("fresh.test".into(), 443))
        .await
        .expect("fresh connect");
    assert_eq!(code, consts::SOCKS5_REPLY_SUCCEEDED);
    raw.stream.write_all(b"again").await.expect("write");
    let mut buf = [0u8; 5];
    raw.stream.read_exact(&mut buf).await.expect("echo");
    assert_eq!(&buf, b"again");

    let (mut assoc, _pump) = raw_associate(&topo).await;
    let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 2], 53)));
    assoc
        .stream
        .write_all(&udp_frame(&target, b"fresh"))
        .await
        .expect("datagram");
    let mut frame_buf = vec![0u8; 2 + 7 + 5];
    assoc
        .stream
        .read_exact(&mut frame_buf)
        .await
        .expect("frame");
    let (_, payload) = udp_unframe(&frame_buf);
    assert_eq!(&payload[..], b"fresh");

    bind_round_trip(&topo, "fresh.test", b"fresh")
        .await
        .expect("fresh bind");
}

/// The teardown matrix, ASSOCIATE row: channel EOF ends the
/// association (the association lifetime = channel lifetime) and
/// fresh sessions of every command succeed immediately after.
#[tokio::test]
async fn teardown_associate_eof_then_every_command_serves_fresh() {
    let topo = full_raw_topology().await;

    let (mut assoc, pump) = raw_associate(&topo).await;
    let target = TargetAddr::Ip(SocketAddr::from(([10, 0, 0, 3], 53)));
    assoc
        .stream
        .write_all(&udp_frame(&target, b"pre"))
        .await
        .expect("datagram");
    let mut frame_buf = vec![0u8; 2 + 7 + 3];
    assoc
        .stream
        .read_exact(&mut frame_buf)
        .await
        .expect("frame");
    let (_, payload) = udp_unframe(&frame_buf);
    assert_eq!(&payload[..], b"pre");

    // Channel EOF ends the association (the clean exit).
    drop(assoc);
    within(pump).await.expect("the stage ends on channel EOF");

    // Fresh sessions of every command succeed immediately.
    let (mut assoc, _pump) = raw_associate(&topo).await;
    assoc
        .stream
        .write_all(&udp_frame(&target, b"fresh"))
        .await
        .expect("datagram");
    let mut frame_buf = vec![0u8; 2 + 7 + 5];
    assoc
        .stream
        .read_exact(&mut frame_buf)
        .await
        .expect("frame");
    let (_, payload) = udp_unframe(&frame_buf);
    assert_eq!(&payload[..], b"fresh");

    connect_round_trip(&topo, "fresh.test".into(), "again".into())
        .await
        .expect("fresh connect");

    bind_round_trip(&topo, "fresh.test", b"fresh")
        .await
        .expect("fresh bind");
}

/// The teardown matrix, BIND row: the client vanishing in the quiet
/// wait releases the accept (the parked accept future drops with the
/// `select!` scope: the client-gone teardown, ADR-004/ADR-007) and
/// fresh sessions succeed immediately after.
#[tokio::test]
async fn teardown_bind_client_gone_releases_the_accept_then_serves_fresh() {
    let topo = wire_with(
        RegisterMode::Producer {
            dial: echo_dial(),
            egress: Some(echo_egress()),
            accept: Some(parked_accept()),
        },
        Some(both_scopes_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    // Reply#1 lands (the parked factory resolves); the quiet wait is
    // then the session's only live phase.
    let (mut raw, pump) = open_adopt_pump(&topo).await;
    raw.greet().await.expect("greet");
    raw.request(
        Socks5Command::TCPBind,
        &TargetAddr::Domain("gone.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);

    // The client vanishes mid-quiet-wait: the session ends cleanly
    // (the accept future is released with it).
    drop(raw);
    within(pump)
        .await
        .expect("the session ends on the client-gone teardown");

    // The accept re-arms: a fresh BIND reaches reply#1 again (and its
    // quiet wait tears down cleanly the same way).
    let (mut raw, pump) = open_adopt_pump(&topo).await;
    raw.greet().await.expect("greet");
    raw.request(
        Socks5Command::TCPBind,
        &TargetAddr::Domain("fresh.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);
    drop(raw);
    within(pump)
        .await
        .expect("the fresh BIND session ends cleanly");

    // CONNECT serves the same topology (no poisoning).
    connect_round_trip(&topo, "fresh.test".into(), "again".into())
        .await
        .expect("fresh connect after the bind teardown");
}

/// The identity/scope matrix, `socks5:open`-only row (ADR-006):
/// CONNECT and ASSOCIATE serve; BIND is refused `ConnectionNotAllowed`
/// (0x02) at the command gate while CONNECT still serves the same
/// identity; ingress is never implied by egress.
#[tokio::test]
async fn identity_matrix_open_only_connect_associate_serve_bind_refused_0x02() {
    let topo = wire_with(
        RegisterMode::Producer {
            dial: echo_dial(),
            egress: Some(echo_egress()),
            accept: Some(echo_accept()),
        },
        Some(consumer_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    // CONNECT serves.
    connect_round_trip(&topo, "open.test".into(), "tcp".into())
        .await
        .expect("CONNECT serves the open-only identity");

    // ASSOCIATE serves.
    associate_round_trip(&topo, 40, b"udp")
        .await
        .expect("ASSOCIATE serves the open-only identity");

    // BIND: the command gate refuses 0x02, RFC-faithful; no data
    // plane (the accept seam never ran).
    let (mut raw, pump) = open_adopt_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,
        "the command gate's 0x02"
    );
    drop(raw);
    within(pump).await.expect("the gated session ends");

    // CONNECT still serves the same identity (the gate distinguishes
    // ingress from egress, not the other way around).
    connect_round_trip(&topo, "still.test".into(), "again".into())
        .await
        .expect("CONNECT still serves after the BIND refusal");

    // The per-call opener identity the gates acted on (CF-006).
    assert_eq!(
        topo.identity_witness.lock().await.as_deref(),
        Some("consumer")
    );
}

/// The identity/scope matrix, `socks5:bind`-only row (ADR-006, as
/// landed): the open op's `required_scopes` is `socks5:open`
/// unconditionally, so a bind-only identity fails closed at the open
/// gate for all three commands (FORBIDDEN; the establisher never
/// runs; no channel anywhere). `socks5:bind` alone grants nothing
/// here; the two grants are never implied, in either direction.
#[tokio::test]
async fn identity_matrix_bind_only_fails_closed_at_the_open_gate() {
    let topo = wire_with(
        RegisterMode::Producer {
            dial: echo_dial(),
            egress: Some(echo_egress()),
            accept: Some(echo_accept()),
        },
        Some(bind_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    // The open itself is refused: fail-closed, no channel anywhere,
    // the establisher never ran (the registry's ACL gate runs first).
    let response = topo
        .consumer_call
        .call(alksocks::OP_SOCKS5_OPEN, Socks5Params::as_json())
        .await;
    let err = response
        .result
        .expect_err("a bind-only identity fails closed");
    assert_eq!(err.code, "FORBIDDEN");
    assert!(topo
        .consumer_manager
        .channel_ids()
        .into_iter()
        .all(|id| id == 0));
    assert!(topo
        .producer
        .manager()
        .channel_ids()
        .into_iter()
        .all(|id| id == 0));
    assert!(
        topo.identity_witness.lock().await.is_none(),
        "the establisher never ran; fail-closed"
    );

    // The same posture through the typed consumer sessions: the BIND
    // session's open is a typed FORBIDDEN open error, no session, no
    // phantom channel.
    let ctopo = wire_consumer_side_with(
        echo_dial(),
        None,
        Some(echo_accept()),
        Some(bind_identity()),
    )
    .await;
    let err = BindSession::bind(
        ctopo.consumer.as_ref(),
        ("bind.test", ECHO_BIND_PORT).to_target_addr().unwrap(),
    )
    .await
    .expect_err("a bind-only identity fails closed");
    assert!(
        expect_forbidden(&err),
        "FORBIDDEN is the open error: {err:?}"
    );
    no_leaked_channels(&ctopo).await;
}

/// The identity/scope matrix, both-scopes row (ADR-006): both grants,
/// granted explicitly (never implied), serve every command on the one
/// registration.
#[tokio::test]
async fn identity_matrix_both_scopes_serve_every_command() {
    let topo = full_raw_topology().await;

    connect_round_trip(&topo, "both.test".into(), "tcp".into())
        .await
        .expect("CONNECT serves the both-scopes identity");
    associate_round_trip(&topo, 50, b"udp")
        .await
        .expect("ASSOCIATE serves the both-scopes identity");
    bind_round_trip(&topo, "both.test", b"bind")
        .await
        .expect("BIND serves the both-scopes identity");

    assert_eq!(
        topo.identity_witness.lock().await.as_deref(),
        Some("consumer-both")
    );
}

/// The readiness contract under concurrent load (the integration
/// level pin; the unit-level test is the producer-associate task's):
/// a hot `Ok(None)` egress across concurrent associations (with a
/// CONNECT interleaved) must not starve the datagram stage's channel
/// branch. Datagrams still egress and echo.
#[tokio::test]
async fn readiness_contract_holds_under_concurrent_load() {
    let topo = wire_with(
        RegisterMode::Producer {
            dial: echo_dial(),
            egress: Some(hot_none_egress()),
            accept: None,
        },
        Some(both_scopes_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    const UDP_TAGS: [&[u8]; 2] = [b"hot-zero0", b"hot-one!!"];
    let mut futs: Vec<RoundTrip> = UDP_TAGS
        .into_iter()
        .enumerate()
        .map(|(i, tag)| udp_fut(&topo, 40 + i as u8, tag))
        .collect();
    futs.push(tcp_fut(&topo, "hot.test".into(), "tcp".into()));
    let results = futures::future::join_all(futs).await;
    for (i, result) in results.into_iter().enumerate() {
        result.unwrap_or_else(|e| panic!("the stage starved under load (session {i}): {e}"));
    }

    // The stage is still healthy: another datagram round-trips.
    associate_round_trip(&topo, 50, b"again")
        .await
        .expect("fresh associate after the hot-egress load");
}

/// The spec'd assembly shape end to end (review 002 D-2): the full
/// `Socks5Producer::builder()` chain (`dial` + `egress` + `accept`)
/// through `.register(...)` (the builder's delegated registration, not
/// the free function), serving all three commands over the harness
/// topology. The one assembly path producer.md §Wiring documents is
/// the one this test pins: a signature drift or a delegation bug in
/// `register` can no longer pass the suite.
#[tokio::test]
async fn producer_builder_registers_and_serves_all_three_commands() {
    let topo = wire_with(
        RegisterMode::ProducerBuilder {
            dial: echo_dial(),
            egress: Some(echo_egress()),
            accept: Some(echo_accept()),
            bind_policy: None,
        },
        Some(both_scopes_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    // CONNECT over the builder's dial seam.
    connect_round_trip(&topo, "builder.test".into(), "tcp".into())
        .await
        .expect("CONNECT serves through the builder registration");

    // ASSOCIATE over the builder's egress seam.
    associate_round_trip(&topo, 60, b"udp")
        .await
        .expect("ASSOCIATE serves through the builder registration");

    // BIND over the builder's accept seam (reply#1 + reply#2 +
    // pass-through: the full phase model).
    bind_round_trip(&topo, "builder.test", b"bind")
        .await
        .expect("BIND serves through the builder registration");

    // The registration is the spec'd one: the establisher rode it
    // (the witness is the builder-delegated registration's probe).
    assert_eq!(
        topo.identity_witness.lock().await.as_deref(),
        Some("consumer-both")
    );
}