alktunnels/tests/producer_bindfirst.rs

//! Integration tests for the bind-first establisher + accept
//! dispatcher (shape 3 — ADR-008 §1): the producer whose establisher
//! BINDS (via the injected `BindFn`), mints the correlation token,
//! contributes `bound` + `listen` to the open reply, and whose pump
//! handler is the accept dispatcher — per accept it opens
//! `channels/tunnel/forwarded` back toward the listen opener and pumps
//! the accepted halves against the adopted channel.
//!
//! Covered (the task's test plan): the full bind-first flow e2e (the
//! acceptance gate), the reply-field projection (both fields seen; a
//! dial open sees neither), EOF teardown (the idle-stream sentinel
//! cancels the loop; the far listener is gone), per-accept failure
//! isolation (a rejected forwarded open drops that accept; the loop
//! serves the next), the typed establishment errors (bind failure →
//! `dial_failed`; registry miss → `unknown_resource`), and the token
//! discipline (two opens mint distinct tokens; each dispatcher stamps
//! ITS OWN plan's token).

mod harness;

use tracing_subscriber::EnvFilter;

fn init_tracing() {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .try_init();
}

use std::sync::Arc;

use alkcall::protocol::wire::CallError;
use alktunnels::params::{parse_listen_reply, Substrate, SubstrateAddr, TunnelParams};
use alktunnels::producer::{
    BindFirstAcceptLoop, BindFn, BoundListener, ResourceRegistry, TargetHandle,
    TunnelEstablishError,
};

use harness::{
    make_forwarded_stub_handler, wire, wire_bindfirst, wire_with, CorrelationMap, RegistrationMode,
    Topology,
};

fn listen_params(resource: &str) -> TunnelParams {
    TunnelParams {
        resource: resource.to_string(),
        substrate: Substrate::Tcp,
    }
}

/// Call the producer's open op on channel 0 with the FULL reply
/// (the consumer's bind-first surface — the reply projection probe).
async fn open_listen_with_reply(
    topo: &Topology,
    params: &TunnelParams,
) -> Result<(u32, serde_json::Value), CallError> {
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": serde_json::json!({
            "resource": params.resource,
            "substrate": "tcp",
        }),
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let out = response.result?;
    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
    Ok((channel_id, out))
}

/// The in-process "accepted connection" the test accept loop yields:
/// a duplex pair, the far end an echo task (the same shape the dial
/// harness echoes with). The peer address is the loop's choice.
fn accepted_halves(peer_addr: (u32, u16)) -> (SubstrateAddr, 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 {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        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);
    (
        SubstrateAddr::HostPort {
            host: peer_addr.0.to_string(),
            port: peer_addr.1,
        },
        TargetHandle {
            read: Box::new(c_read),
            write: Box::new(c_write),
        },
    )
}

/// The test accept loop's feed point (the test plays the app server):
/// push an accepted pair; the dispatcher's loop picks it up.
async fn push_accept(
    queue: &Arc<tokio::sync::Mutex<std::collections::VecDeque<(SubstrateAddr, TargetHandle)>>>,
    notify: &Arc<tokio::sync::Notify>,
    peer: (u32, u16),
) {
    let (peer_addr, handle) = accepted_halves(peer);
    queue.lock().await.push_back((peer_addr, handle));
    notify.notify_waiters();
}

/// The assembly's bind: produces a "listener" (the queue's identity)
/// and the OS-chosen bound address (the test's fake — a real
/// `bind_tcp` variant lives in the `local` feature).
fn test_bind() -> BindFn {
    Arc::new(|_substrate, _backing| {
        Box::pin(async {
            Ok(BoundListener {
                listener: Arc::new(4242u32) as Arc<dyn std::any::Any + Send + Sync>,
                bound: SubstrateAddr::HostPort {
                    host: "127.0.0.1".to_string(),
                    port: 42113,
                },
            })
        })
    })
}

/// A failing bind (the `dial_failed`-class probe — the bind IS the
/// dial here).
fn failing_bind(message: &'static str) -> BindFn {
    Arc::new(move |_substrate, _backing| {
        Box::pin(async move { Err(TunnelEstablishError::DialFailed(message.to_string())) })
    })
}

fn only_channel_0(topo: &Topology) -> bool {
    topo.consumer_manager
        .channel_ids()
        .iter()
        .all(|&id| id == 0)
        && topo
            .producer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
}

/// The bind-first topology keeps the LISTEN channel open (the
/// dispatcher holds it) — the "no stray channels" probe excludes it
/// (and channel 0, the call plane) on both sides.
fn no_stray_channels(topo: &Topology, listen_id: u32) -> bool {
    topo.consumer_manager
        .channel_ids()
        .iter()
        .all(|&id| id == 0 || id == listen_id)
        && topo
            .producer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0 || id == listen_id)
}

/// The forwarded-op serving stub on the opener (consumer) side +
/// its correlation map (the assembly-side bookkeeping the real
/// opener-side registration owns).
async fn install_forwarded_stub(topo: &Topology) -> CorrelationMap {
    let correlations: CorrelationMap = Arc::new(tokio::sync::Mutex::new(Default::default()));
    let core = alkcall::channels::operations::ChannelCore::new(
        topo.consumer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    core.register_openable(
        alktunnels::params::tunnel_forwarded_spec(),
        make_forwarded_stub_handler(Arc::clone(&correlations)),
        &topo.consumer_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
    )
    .expect("register forwarded stub on the opener");
    correlations
}

#[tokio::test]
async fn bindfirst_full_flow_end_to_end() {
    init_tracing();
    // THE acceptance gate (ADR-008): opener opens `channels/tunnel/sub`
    // with the bind-first producer registered → the reply carries
    // `bound` + `listen` → the opener serves `channels/tunnel/forwarded`
    // with a correlation map keyed on the token → the test accept loop
    // accepts → the producer's dispatcher opens forwarded with
    // `{listen: token, substrate, peer}` → the opener correlates,
    // pumps against the local BIND client halves → byte round-trip
    // through the forwarded channel.
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "assembly-owned")
        .await;
    let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new()));
    let notify = Arc::new(tokio::sync::Notify::new());
    let topo = wire_bindfirst(
        registry,
        test_bind(),
        harness_queue_loop(queue.clone(), notify.clone()),
    )
    .await;

    // The opener opens the listen tunnel (with-reply surface — the
    // bind-first reply projection).
    let (channel_id, reply) = open_listen_with_reply(&topo, &listen_params("bind"))
        .await
        .expect("listen open");

    // Reply-field projection: `bound` + `listen` ride the success
    // output alongside the wrapper's `channel_id`.
    let projected = parse_listen_reply(&reply).expect("projected reply");
    assert_eq!(
        projected.bound,
        Some(SubstrateAddr::HostPort {
            host: "127.0.0.1".to_string(),
            port: 42113,
        }),
        "the `bound` field is the OS-chosen (test-chosen) address"
    );
    let token = projected.listen.expect("the `listen` token");
    assert!(token > 0);

    // The opener correlates on the token (the assembly-side map).
    let correlations = install_forwarded_stub(&topo).await;

    // The LOCAL BIND client (the app-server-side session on the
    // opener): a raw duplex pair — the map's halves are the stub's
    // pump target, the far end is the test's client end. The stub's
    // `pump_bidi` bridges the forwarded channel ↔ these halves, so
    // bytes the test writes on the far end traverse the whole path
    // (local BIND client ↔ opener stub pump ↔ forwarded channel ↔
    // producer-side pump ↔ accepted handle's echo task) and echo back.
    let (client_read, client_write, l_read, l_write) = bind_client_pair();
    correlations.lock().await.insert(
        token,
        TargetHandle {
            read: Box::new(l_read),
            write: Box::new(l_write),
        },
    );

    // The opener adopts the LISTEN channel (the idle side — its EOF
    // is the teardown signal).
    topo.consumer_manager
        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen channel");

    // The test accept loop accepts (the app server dialed).
    push_accept(&queue, &notify, (198, 55231)).await;

    // The dispatcher opens forwarded; the opener's stub handler
    // correlates by token and pumps against the local halves. The
    // forwarded channel ID arrives on the opener's manager — find it
    // (the only non-0, non-listen channel on the opener side after
    // the adopt + forwarded open settle).
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    let forwarded_ids: Vec<u32> = topo
        .consumer_manager
        .channel_ids()
        .into_iter()
        .filter(|&id| id != 0 && id != channel_id)
        .collect();
    assert_eq!(forwarded_ids.len(), 1, "one forwarded channel arrived");

    // The round-trip: bytes the test writes on the local BIND
    // client's end cross the stub's pump → the forwarded channel →
    // the producer's per-accept pump → the accepted handle's echo →
    // and back (the full bind-first data path, ADR-008's gate).
    let (mut client_read, mut client_write) = (client_read, client_write);
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    client_write.write_all(b"bind-ping").await.expect("write");
    client_write.flush().await.expect("flush");
    let mut buf = vec![0u8; 9];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        client_read.read_exact(&mut buf),
    )
    .await
    .expect("round trip")
    .expect("read");
    assert_eq!(&buf, b"bind-ping");

    // Teardown follows the accepted side: the far echo closes when the
    // accepted handle's pump sees the test client's EOF. (No channel
    // assertions here — the listen channel legitimately stays open.)
}

/// The listen-channel EOF teardown: drop the listener session (close
/// the listen channel's stream) → the accept loop cancels, the far
/// listener is gone (a follow-on accept attempt fails), and the
/// forwarded channels already adopted own their teardown
/// independently.
#[tokio::test]
async fn listen_channel_eof_cancels_loop_and_tears_listener_down() {
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "assembly-owned")
        .await;
    let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new()));
    let notify = Arc::new(tokio::sync::Notify::new());
    let topo = wire_bindfirst(
        registry,
        test_bind(),
        harness_queue_loop(queue.clone(), notify.clone()),
    )
    .await;

    let (channel_id, reply) = open_listen_with_reply(&topo, &listen_params("bind"))
        .await
        .expect("listen open");
    let token = parse_listen_reply(&reply)
        .expect("projected")
        .listen
        .expect("token");

    // The opener adopts the listen channel and then DROPS its stream
    // (the listener session is dropped — the zero-byte EOF sentinel).
    let (send, _recv) = topo
        .consumer_manager
        .adopt_channel(channel_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen channel");
    drop(send);

    // The opener-side teardown reaps the listen channel.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(
        !topo.producer.manager().has_channel(channel_id),
        "the listen channel reaped after the session dropped"
    );

    // A follow-on accept attempt: the loop is gone — the accept queue
    // accepts nothing new (the listener was torn down with the plan).
    // Observable via the queue: nothing pumps it anymore.
    let _ = token;
    push_accept(&queue, &notify, (198, 55232)).await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert_eq!(
        queue.lock().await.len(),
        1,
        "the accept loop is gone: the pushed accept was never consumed"
    );
}

/// Per-accept failure isolation: the opener's forwarded registry
/// rejects one forwarded open (scope-less identity) → that accept's
/// connection drops, the loop serves the next accept.
#[tokio::test]
async fn forwarded_open_failure_isolates_the_accept() {
    init_tracing();
    // The topology's consumer identity holds `tunnel:open` but NOT
    // `tunnel:forwarded` — every forwarded open fails FORBIDDEN at the
    // opener's ACL. The loop must keep serving (a second accept
    // produces a second forwarded-open attempt).
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "assembly-owned")
        .await;
    let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new()));
    let notify = Arc::new(tokio::sync::Notify::new());
    let topo = wire_with(
        registry,
        RegistrationMode::BindFirst(
            test_bind(),
            harness_queue_loop(queue.clone(), notify.clone()),
        ),
        Some(harness::forwarded_scopeless_identity()),
        None,
        Arc::new(alkcall::core::auth::NoopIdentityProvider),
    )
    .await;

    let (listen_id, _reply) = open_listen_with_reply(&topo, &listen_params("bind"))
        .await
        .expect("listen open");

    // First accept: the forwarded open fails (scope-less opener) —
    // the accepted connection drops, the loop continues.
    push_accept(&queue, &notify, (198, 55233)).await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert!(
        no_stray_channels(&topo, listen_id),
        "a failed forwarded open leaves no channel beyond the listen channel"
    );

    // Second accept: the loop is still alive (per-accept isolation) —
    // this one fails the same way, but the LOOP's health is what the
    // test pins: the accept was CONSUMED (the queue drained).
    push_accept(&queue, &notify, (198, 55234)).await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    let remaining = queue.lock().await.len();
    assert_eq!(
        remaining, 0,
        "the second accept was consumed too (the loop served it)"
    );
    assert!(no_stray_channels(&topo, listen_id));
}

/// Establishment error paths: bind failure → `dial_failed`-class on
/// the wire (`channel:open_failed`); registry miss →
/// `unknown_resource`.
#[tokio::test]
async fn establishment_errors_are_typed_on_the_wire() {
    // Bind failure → dial_failed (the bind IS the dial here).
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "assembly-owned")
        .await;
    let topo = wire_bindfirst(
        registry,
        failing_bind("address already in use"),
        noop_loop(),
    )
    .await;
    match open_listen_with_reply(&topo, &listen_params("bind")).await {
        Ok(_) => panic!("bind failure must fail the open"),
        Err(e) => {
            assert_eq!(e.code, "channel:open_failed");
            let reason = e
                .details
                .as_ref()
                .and_then(|d| d.get("reason"))
                .and_then(|r| r.as_str())
                .expect("reason in details");
            assert_eq!(reason, "dial_failed");
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    assert!(only_channel_0(&topo), "a failed open leaves no channel");

    // Registry miss → unknown_resource.
    let registry = ResourceRegistry::new();
    let topo = wire_bindfirst(registry, test_bind(), noop_loop()).await;
    match open_listen_with_reply(&topo, &listen_params("ghost")).await {
        Ok(_) => panic!("registry miss must fail the open"),
        Err(e) => {
            assert_eq!(e.code, "channel:open_failed");
            let reason = e
                .details
                .as_ref()
                .and_then(|d| d.get("reason"))
                .and_then(|r| r.as_str())
                .expect("reason in details");
            assert_eq!(reason, "unknown_resource");
        }
    }
    tokio::time::sleep(std::time::Duration::from_millis(150)).await;
    assert!(only_channel_0(&topo));
}

/// Token discipline: two opens on the same producer mint distinct
/// tokens; each dispatcher stamps ITS OWN plan's token (the second
/// open's forwarded opens carry the second token — pinned via the
/// correlation stub's map: the FIRST token's entry is left
/// unconsumed).
#[tokio::test]
async fn token_discipline_two_opens_distinct_tokens() {
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Tcp, "assembly-owned")
        .await;
    let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new()));
    let notify = Arc::new(tokio::sync::Notify::new());
    let topo = wire_bindfirst(
        registry,
        test_bind(),
        harness_queue_loop(queue.clone(), notify.clone()),
    )
    .await;
    let correlations = install_forwarded_stub(&topo).await;

    let (id1, reply1) = open_listen_with_reply(&topo, &listen_params("bind"))
        .await
        .expect("open 1");
    let (id2, reply2) = open_listen_with_reply(&topo, &listen_params("bind"))
        .await
        .expect("open 2");
    assert_ne!(id1, id2);
    let t1 = parse_listen_reply(&reply1)
        .expect("reply 1")
        .listen
        .expect("token 1");
    let t2 = parse_listen_reply(&reply2)
        .expect("reply 2")
        .listen
        .expect("token 2");
    assert_ne!(t1, t2, "two opens mint distinct tokens");

    // Adopt both listen channels (both sessions live).
    topo.consumer_manager
        .adopt_channel(id1, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen 1");
    topo.consumer_manager
        .adopt_channel(id2, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen 2");

    // Both dispatchers are live. Correlate BOTH tokens (each opener
    // session keys its own), but only register the local halves for
    // t1 — a forwarded open stamped t2 would correlation-miss. Push
    // one accept AFTER open 2: whichever dispatcher consumed it, the
    // STAMPED token is the plan's own. To pin "each dispatcher stamps
    // ITS OWN plan's token", correlate both but track which token the
    // forwarded open carried.
    correlations.lock().await.insert(t1, {
        let (l_read, l_write, _, _) = bind_client_pair();
        TargetHandle {
            read: Box::new(l_read),
            write: Box::new(l_write),
        }
    });
    correlations.lock().await.insert(t2, {
        let (l_read, l_write, _, _) = bind_client_pair();
        TargetHandle {
            read: Box::new(l_read),
            write: Box::new(l_write),
        }
    });

    push_accept(&queue, &notify, (198, 55235)).await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    // The accept was consumed by exactly one dispatcher, and the
    // forwarded open succeeded (correlated by its stamped token).
    assert_eq!(queue.lock().await.len(), 0, "one accept consumed");
    assert_eq!(
        correlations.lock().await.len(),
        1,
        "one token's halves were consumed"
    );
}

/// The forwarded-op serving stub's forwarded-substrate log: the
/// `substrate` param of every forwarded open (the M-1 remediation
/// probe surface — the substrate's SOURCE: the plan, not a re-parse).
pub type SubstrateLog = Arc<tokio::sync::Mutex<Vec<alktunnels::params::Substrate>>>;

/// The forwarded-op serving stub variant that records each forwarded
/// open's `substrate` param, then pumps against the correlated halves
/// (the same shape as `make_forwarded_stub_handler` + the log).
fn make_substrate_recording_stub_handler(
    correlations: CorrelationMap,
    substrates: SubstrateLog,
) -> alkcall::channels::operations::OpenHandler {
    Arc::new(
        move |input: serde_json::Value,
              _plan: Option<alkcall::channels::operations::ChannelPlan>,
              conn: alkcall::core::types::Connection,
              _auth: alkcall::core::auth::AuthContext| {
            let correlations = Arc::clone(&correlations);
            let substrates = Arc::clone(&substrates);
            tokio::spawn(async move {
                let forwarded: alktunnels::params::TunnelForwardedParams =
                    match serde_json::from_value(input) {
                        Ok(p) => p,
                        Err(e) => {
                            tracing::error!(error = %e, "substrate stub: params parse failed");
                            return;
                        }
                    };
                substrates.lock().await.push(forwarded.substrate);
                let target = correlations.lock().await.remove(&forwarded.listen);
                let Some(target) = target else {
                    tracing::error!(
                        listen = forwarded.listen,
                        "substrate stub: correlation miss"
                    );
                    return;
                };
                let Some(bidi) = conn.accept_bi().await.ok() else {
                    tracing::error!("substrate stub: accept_bi failed");
                    return;
                };
                alkcall::channels::pump::pump_bidi(bidi, target.read, target.write).await;
            })
        },
    )
}

/// The forwarded-op serving stub recording substrates, installed on
/// the opener (consumer) side (the same post-hoc registration posture
/// as `install_forwarded_stub`).
async fn install_substrate_recording_stub(
    topo: &Topology,
    substrates: SubstrateLog,
) -> CorrelationMap {
    let correlations: CorrelationMap = Arc::new(tokio::sync::Mutex::new(Default::default()));
    let core = alkcall::channels::operations::ChannelCore::new(
        topo.consumer_manager.clone(),
        alkcall::channels::policy::default_policy(),
    );
    core.register_openable(
        alktunnels::params::tunnel_forwarded_spec(),
        make_substrate_recording_stub_handler(Arc::clone(&correlations), substrates),
        &topo.consumer_registry,
        alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
    )
    .expect("register substrate-recording forwarded stub on the opener");
    correlations
}

/// M-1 remediation pin: the forwarded opens' `substrate` comes from
/// the PLAN (stamped at establisher time), not a re-parse of wire
/// input — a UDP bind-first listen produces forwarded opens carrying
/// `substrate: udp` even though the re-parse-and-fallback shape would
/// have stamped `tcp` if it ever fell back. (The fallback itself is
/// REMOVED — this test pins the plan as the source by observing the
/// forwarded params through a real dispatcher round trip.)
#[tokio::test]
async fn forwarded_opens_carry_plan_stamped_substrate() {
    init_tracing();
    // A UDP-flavored bind-first registry: the open is
    // `{resource: "bind", substrate: "udp"}` — the plan must stamp
    // Udp (the old fallback shape would have stamped Tcp on any
    // parse failure).
    let registry = ResourceRegistry::new();
    registry
        .register("bind", Substrate::Udp, "assembly-owned")
        .await;
    let queue = Arc::new(tokio::sync::Mutex::new(std::collections::VecDeque::new()));
    let notify = Arc::new(tokio::sync::Notify::new());
    let topo = wire_bindfirst(
        registry,
        test_bind(),
        harness_queue_loop(queue.clone(), notify.clone()),
    )
    .await;

    let substrates: SubstrateLog = Arc::new(tokio::sync::Mutex::new(Vec::new()));
    let correlations = install_substrate_recording_stub(&topo, Arc::clone(&substrates)).await;

    // The listen open over the UDP substrate (the harness's
    // open-call helper hardcodes "tcp" — call with the udp input
    // directly; the schema gate accepts the registered UDP resource).
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": {"resource": "bind", "substrate": "udp"},
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let reply = response.result.expect("udp listen open");
    let listen_id = reply["channel_id"].as_u64().expect("channel_id") as u32;
    let token = parse_listen_reply(&reply)
        .expect("projected reply")
        .listen
        .expect("the `listen` token");

    // The opener correlates on the token and adopts the listen
    // channel (the idle side).
    let (l_read, l_write, _, _) = bind_client_pair();
    correlations.lock().await.insert(
        token,
        TargetHandle {
            read: Box::new(l_read),
            write: Box::new(l_write),
        },
    );
    topo.consumer_manager
        .adopt_channel(listen_id, alktunnels::TUNNEL_ALPN, None)
        .await
        .expect("adopt listen channel");

    // One accept: the dispatcher opens forwarded — the forwarded
    // params' `substrate` must be the PLAN's (Udp), not a fallback.
    push_accept(&queue, &notify, (198, 55236)).await;
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;

    let logged = substrates.lock().await.clone();
    assert_eq!(
        logged,
        vec![alktunnels::params::Substrate::Udp],
        "the forwarded open's substrate is plan-stamped (Udp), not re-parsed"
    );
}

/// A plain dial open sees NEITHER reply field (the projection
/// property — the reserved-key guard upstream guarantees `channel_id`
/// stays wrapper-owned; a dial establisher contributes nothing).
#[tokio::test]
async fn dial_open_reply_has_neither_field() {
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, harness::echo_dial()).await;
    let payload = serde_json::json!({
        "operationId": "channels/tunnel/sub",
        "input": {"resource": "echo", "substrate": "tcp"},
    });
    let response = topo.consumer_call.call_with_payload(payload).await;
    let out = response.result.expect("dial open");
    let projected = parse_listen_reply(&out).expect("projected");
    assert_eq!(
        projected,
        alktunnels::params::TunnelListenReply {
            bound: None,
            listen: None
        },
        "a dial establisher's open carries neither field"
    );
}

/// The accept loop's queue-shaped stand-in (the harness_queue_loop in
/// the tests above): a closure popping from the test's VecDeque +
/// Notify pair. (Named separately from the generic helper so the
/// tests read: queue in, loop out.)
fn harness_queue_loop(
    queue: Arc<tokio::sync::Mutex<std::collections::VecDeque<(SubstrateAddr, TargetHandle)>>>,
    notify: Arc<tokio::sync::Notify>,
) -> BindFirstAcceptLoop {
    Arc::new(move |_listener| {
        let queue = Arc::clone(&queue);
        let notify = Arc::clone(&notify);
        Box::pin(async move {
            // The queue-driven accept: check once, wait for a push,
            // check again (the Notify lost-wakeup pattern — the
            // `Notified` future is created BEFORE the state check on
            // the waiting path). One accepted pair per closure call
            // (the dispatcher invokes the closure per accept).
            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()
                }
            };
            match popped {
                Some(pair) => Ok(pair),
                None => Err(TunnelEstablishError::DialFailed(
                    "listener closed".to_string(),
                )),
            }
        })
    })
}

/// A never-accepting loop (the establishment-error probes — the loop
/// never runs: the open fails at the establisher).
fn noop_loop() -> BindFirstAcceptLoop {
    Arc::new(|_listener| {
        Box::pin(async { Err(TunnelEstablishError::DialFailed("noop loop".to_string())) })
    })
}

/// The "local BIND client" pair (the opener-side pump target): a raw
/// duplex pair — the map's halves are the stub's pump target, the far
/// end is the test's client end (driven directly; the accepted side's
/// echo task closes the loop).
fn bind_client_pair() -> (
    tokio::io::ReadHalf<tokio::io::DuplexStream>,
    tokio::io::WriteHalf<tokio::io::DuplexStream>,
    tokio::io::ReadHalf<tokio::io::DuplexStream>,
    tokio::io::WriteHalf<tokio::io::DuplexStream>,
) {
    let (a, b) = tokio::io::duplex(64 * 1024);
    let (a_read, a_write) = tokio::io::split(a);
    let (b_read, b_write) = tokio::io::split(b);
    (b_read, b_write, a_read, a_write)
}