alktunnels/tests/consumer_session.rs

//! Integration tests for the consumer half — `TunnelSession` per
//! consumer.md + ADR-005, over the duplex harness. Ported from the
//! reverse POC's `ReverseTunnel` suite (the W3/W4 probes) plus the
//! forward POC's session shapes, generalized to the spec's session
//! type.
//!
//! Covered: forward open (session halves drive a duplex round-trip),
//! reverse open+adopt+pump_against (round-trip, W4 half-close, join
//! copy counts, out-of-band close + self-reaping, pump-less join),
//! datagram variant (round-trip incl. the empty datagram — the F-2
//! layering), wrong-substrate errors, and the teardown matrix
//! (close/join/Drop — no leaked channel entries via `channel_ids()`).

mod harness;

use alktunnels::params::{Substrate, TunnelParams};
use alktunnels::producer::ResourceRegistry;
use alktunnels::{open_reverse_channel, TunnelSession};

use harness::{
    all_substrate_dial, closing_udp_echo_dial, echo_dial, framed_udp_echo_dial, wire, wire_forward,
    Topology,
};

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

/// Both managers hold only channel 0 (the phantom-channel property,
/// asserted from both topologies).
trait LeakCheck {
    fn no_data_channels(&self) -> bool;
}

impl LeakCheck for Topology {
    fn no_data_channels(&self) -> bool {
        self.consumer_manager
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
            && self
                .producer
                .manager()
                .channel_ids()
                .iter()
                .all(|&id| id == 0)
    }
}

impl LeakCheck for harness::ForwardTopology {
    fn no_data_channels(&self) -> bool {
        self.consumer
            .manager()
            .channel_ids()
            .iter()
            .all(|&id| id == 0)
            && self
                .producer_manager
                .channel_ids()
                .iter()
                .all(|&id| id == 0)
    }
}

fn only_channel_0<T: LeakCheck>(topo: &T) -> bool {
    topo.no_data_channels()
}

#[tokio::test]
async fn forward_open_session_halves_round_trip() {
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire_forward(registry, echo_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("open");
    assert!(topo.producer_manager.has_channel(session.channel_id));

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (read, write) = session.stream_halves().expect("stream session has halves");
    write.write_all(b"ping").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 4];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip")
        .expect("read");
    assert_eq!(&buf, b"ping");

    // Teardown: the session reaps its adopted entry (ADR-005).
    let channel_id = session.channel_id;
    let reaped = session.close().await;
    assert!(reaped, "close reaped the adopted entry");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    assert!(!topo.producer_manager.has_channel(channel_id));
    assert!(only_channel_0(&topo), "no leaked entries after close");
}

#[tokio::test]
async fn forward_open_take_halves_then_pumpless_join() {
    // consumer.md's pinned semantics: after take_halves the session is
    // pump-less — join completes immediately, reaps only, (0, 0, reaped).
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire_forward(registry, echo_dial()).await;

    let session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("open");
    let taken = session.take_halves();

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut read = taken.read;
    let mut write = taken.write;
    write.write_all(b"echo-me").await.expect("write");
    write.flush().await.expect("flush");
    let mut buf = vec![0u8; 7];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("round trip through taken halves")
        .expect("read");
    assert_eq!(&buf, b"echo-me");

    let session = taken.session;
    // Drop the halves BEFORE joining: EOF propagates through the mux,
    // the producer-side wrapper reaps, and the pump-less session reaps
    // its own adopted entry on join (nothing to await).
    drop(read);
    drop(write);
    let (c2p, p2c, reaped) = session.join().await;
    assert_eq!((c2p, p2c), (0, 0), "pump-less join: no pump to await");
    assert!(reaped, "pump-less join reaps the adopted entry");
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn datagram_round_trip_including_empty() {
    // The datagram variant over the codec (ADR-003): round-trip incl.
    // the F-2 empty datagram — `len=0` is a legal datagram, never EOF.
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, framed_udp_echo_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    assert!(
        session.stream_halves().is_none(),
        "stream_halves is stream-substrate-only"
    );

    session
        .send_datagram(b"query-1")
        .await
        .expect("send datagram");
    session
        .send_datagram(b"")
        .await
        .expect("send empty datagram");

    let dg1 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("no io error")
        .expect("first datagram");
    assert_eq!(&dg1[..], b"query-1");
    let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("no io error")
        .expect("second datagram (empty round-trips — F-2)");
    assert!(dg2.is_empty(), "the empty datagram round-trips (F-2)");

    let reaped = session.close().await;
    assert!(reaped);
    // The producer-side wrapper reap is async (the pump task exits,
    // the wrapper reaps) — settle before asserting both managers.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn datagram_backlog_tracks_pending_and_drains_fifo() {
    // The §1.3/§3.3 session-shape diagnostic: `datagram_backlog()` is
    // the `DatagramHalves.pending` count — None for stream sessions,
    // 0 at rest, batched arrivals queue FIFO, and a drained queue
    // reads 0 (the wave-2 harness asserts the bound against it).
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire_forward(registry, all_substrate_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    assert_eq!(
        session.datagram_backlog(),
        Some(0),
        "a fresh datagram session has an empty backlog"
    );
    let stream_session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Tcp)).await;
    // This registry entry maps (dns, Tcp) nowhere; open fails typed.
    assert!(stream_session.is_err(), "unproduced substrate rejects");
    let stream_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("tcp open");
    assert_eq!(
        stream_session.datagram_backlog(),
        None,
        "stream sessions carry no datagram backlog"
    );
    stream_session.close().await;

    // Backlog drain posture: two echoed datagrams land back-to-back;
    // whether the echo batches them into one chunk is timing-shaped,
    // so the suite pins only the rest state (empty after drains) —
    // the batched-arrival bound is the wave-2 harness's (§3.3).
    session.send_datagram(b"first").await.expect("send");
    session.send_datagram(b"second").await.expect("send");
    let dg1 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("no io error")
        .expect("first datagram");
    let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("no io error")
        .expect("second datagram");
    let mut got = [&dg1[..], &dg2[..]];
    got.sort();
    assert_eq!(got, [b"first".as_slice(), b"second".as_slice()]);
    assert_eq!(
        session.datagram_backlog(),
        Some(0),
        "the backlog drains to empty once all datagrams are pulled"
    );

    let reaped = session.close().await;
    assert!(reaped);
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn wrong_substrate_operations_are_typed() {
    // The POC's WrongSubstrate shape: substrate-shaped operations fail
    // typed on the wrong plane.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, all_substrate_dial()).await;

    let mut stream_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("tcp open");
    let err = stream_session
        .send_datagram(b"nope")
        .await
        .expect_err("send_datagram on a stream session");
    assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate));
    let err = stream_session
        .recv_datagram()
        .await
        .expect_err("recv_datagram on a stream session");
    assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate));
    stream_session.close().await;

    let mut dg_session =
        match TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)).await {
            Err(_) => panic!("setup: udp dial must succeed in the framed harness"),
            Ok(mut s) => {
                assert!(
                    s.stream_halves().is_none(),
                    "stream_halves is stream-substrate-only"
                );
                s
            }
        };
    dg_session.send_datagram(b"late").await.expect("send works");
    dg_session.close().await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn reverse_open_adopt_pump_round_trip() {
    // The reverse flow (the POC's ReverseTunnel shape, generalized):
    // open_reverse_channel → adopt → pump_against → round-trip →
    // join (copy counts + reap).
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;

    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
            .await
            .expect("reverse open call");

    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");
    assert_eq!(session.channel_id, channel_id);

    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
    let session = session.pump_against(accepted_end).await;

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (mut l_read, mut l_write) = tokio::io::split(local_end);
    l_write.write_all(b"reverse").await.expect("write");
    l_write.flush().await.expect("flush");
    let mut buf = vec![0u8; 7];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        l_read.read_exact(&mut buf),
    )
    .await
    .expect("round trip")
    .expect("read");
    assert_eq!(&buf, b"reverse");

    // EOF propagates through the two pumps (the two-pump contract).
    drop(l_write);
    drop(l_read);
    let (c2p, p2c, reaped) =
        tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
            .await
            .expect("join timed out");
    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
    assert!(reaped, "join reaps after pump completion");
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    assert!(
        !topo.producer.manager().has_channel(channel_id),
        "producer-side channel reaped (wrapper-managed)"
    );
    assert!(only_channel_0(&topo), "no leaked entries after join");
}

#[tokio::test]
async fn reverse_half_close_semantics_w4() {
    // W4's shape: the local side half-closes (shutdown write); the
    // target sees EOF on its read side while the reverse direction
    // stays pumpable, and the target's final reply still flows back.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;

    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
            .await
            .expect("reverse open call");
    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");

    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
    let session = session.pump_against(accepted_end).await;

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let (mut l_read, mut l_write) = tokio::io::split(local_end);
    l_write.write_all(b"half").await.expect("write");
    l_write.flush().await.expect("flush");
    l_write.shutdown().await.expect("half close");

    let mut buf = vec![0u8; 4];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        l_read.read_exact(&mut buf),
    )
    .await
    .expect("echo after half close")
    .expect("read echo");
    assert_eq!(&buf, b"half");

    let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
        .await
        .expect("join after half close");
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn out_of_band_close_then_session_self_reap() {
    // The POC's reverse_channel_close_out_of_band shape: a peer-initiated
    // `channel/close` tears the SERVING side down; the consumer's session
    // still owns its adopted entry — close/join reaps it (W3 split).
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;

    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
            .await
            .expect("reverse open call");
    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");
    let session = session.pump_against(tokio::io::duplex(1024).0).await;

    let response = topo
        .consumer_call
        .call_with_payload(serde_json::json!({
            "operationId": "channel/close",
            "input": { "channel_id": channel_id, "reason": "out-of-band" }
        }))
        .await;
    assert!(response.result.is_ok(), "channel/close resolved");

    // The close ran on the PRODUCER side; the consumer's entry remains.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(
        !topo.producer.manager().has_channel(channel_id),
        "producer-side channel reaped by the out-of-band close"
    );
    assert!(topo.consumer_manager.has_channel(channel_id));
    let reaped = session.close().await;
    assert!(reaped, "session reaped its adopted entry");
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn drop_never_leaks_the_channel_entry() {
    // ADR-005: Drop = abort + sync reap (best-effort). Dropping a live
    // session without close/join never leaks the entry.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let ftopo = wire_forward(registry.clone(), echo_dial()).await;

    let session = TunnelSession::open(&ftopo.consumer, params("echo", Substrate::Tcp))
        .await
        .expect("open");
    let channel_id = session.channel_id;
    assert!(ftopo.consumer.manager().has_channel(channel_id));
    drop(session);
    assert!(
        !ftopo.consumer.manager().has_channel(channel_id),
        "Drop reaped the adopted entry"
    );

    // Drop with a session-owned pump: the pump aborts too (the reverse
    // topology — pump_against is the reverse-flow use).
    let topo = wire(registry, echo_dial()).await;
    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
            .await
            .expect("second reverse open");
    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");
    let session = session.pump_against(tokio::io::duplex(1024).0).await;
    assert!(topo.consumer_manager.has_channel(channel_id));
    drop(session);
    assert!(!topo.consumer_manager.has_channel(channel_id));
}

#[tokio::test]
async fn oversize_datagram_rejected_at_frame_time() {
    // ADR-003: a >65535-byte send is rejected at frame time (`Oversize`),
    // never a wire overflow.
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, framed_udp_echo_dial()).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    let err = session
        .send_datagram(&vec![0u8; alktunnels::MAX_DATAGRAM_LEN + 1])
        .await
        .expect_err("oversize must fail");
    assert!(
        matches!(
            err,
            alktunnels::TunnelIoError::Codec(alktunnels::wire::DatagramCodecError::Oversize(_, _))
        ),
        "oversize is a codec error at frame time: {err:?}"
    );
    session.close().await;
}

#[tokio::test]
async fn datagram_clean_eof_is_none_not_error() {
    // consumer.md's EOF contract for pump-less datagram sessions: a
    // clean stream EOF (the channels-level sentinel, mid-frame
    // state clear) resolves `Ok(None)` — the caller's shutdown
    // signal, distinct from both an empty datagram (`Some(b"")`,
    // F-2) and a truncated stream (`TruncatedDatagram`).
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, closing_udp_echo_dial(true)).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    session.send_datagram(b"last").await.expect("send");

    // The echo's far end drops after replying: the channel's read
    // half EOFs AFTER the datagram (clean boundary).
    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect("io ok")
        .expect("the echoed datagram");
    assert_eq!(&dg[..], b"last");

    let eof = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("eof recv timed out")
        .expect("clean EOF is not an io error");
    assert!(eof.is_none(), "clean EOF is Ok(None), got {eof:?}");

    let (_, _, reaped) = session.join().await;
    assert!(reaped);
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn datagram_mid_frame_eof_fails_loud_truncated() {
    // The fail-loud posture (OQ-TN-13): a stream end mid-datagram
    // (declared length not satisfied) is `TruncatedDatagram`, never a
    // silent partial datagram and never a clean `None`.
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, closing_udp_echo_dial(false)).await;

    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    session.send_datagram(b"cut").await.expect("send");

    // The echo's far end forwards the frame but drops the connection
    // BEFORE the payload completes (the 2-byte length prefix rides;
    // the payload is cut — mid-datagram EOF).
    let err = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
        .await
        .expect("recv timed out")
        .expect_err("mid-datagram EOF must fail loud");
    assert!(
        matches!(err, alktunnels::TunnelIoError::TruncatedDatagram),
        "mid-frame EOF is TruncatedDatagram, got {err:?}"
    );

    session.close().await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn datagram_data_plane_after_take_halves_is_channel_taken() {
    // After the halves were taken, the data plane is gone: datagram
    // accessors fail with `ChannelTaken` (the pinned `ChannelTaken`
    // surface — the session is still the channel-entry owner).
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire_forward(registry, framed_udp_echo_dial()).await;

    let session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
        .await
        .expect("udp open");
    let taken = session.take_halves();
    let alktunnels::TakenHalves {
        session: mut taken_session,
        mut read,
        mut write,
    } = taken;

    let err = taken_session
        .send_datagram(b"late")
        .await
        .expect_err("send after take");
    assert!(matches!(err, alktunnels::TunnelIoError::ChannelTaken));
    let err = taken_session
        .recv_datagram()
        .await
        .expect_err("recv after take");
    assert!(matches!(err, alktunnels::TunnelIoError::ChannelTaken));

    // The taken halves carry the WIRE framing (ADR-005's N-3 pin):
    // the raw bytes are `[len: u16 BE][payload]` — framing duty
    // transferred to the caller, visible on the wire.
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    write
        .write_all(&[0, 3, b'a', b'b', b'c'])
        .await
        .expect("write a raw frame");
    let mut buf = vec![0u8; 5];
    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
        .await
        .expect("raw frame round trip")
        .expect("read");
    assert_eq!(&buf, &[0, 3, b'a', b'b', b'c']);

    // Drop the halves: both pump directions EOF on the producer side
    // (drop of the channel send/recv streams IS the EOF propagation).
    drop(read);
    drop(write);
    let (c2p, p2c, reaped) = taken_session.join().await;
    assert_eq!((c2p, p2c), (0, 0), "pump-less join after take_halves");
    assert!(reaped);
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn reverse_udp_pump_against_framed_adapter_round_trips() {
    // The reverse UDP shape: the session's codec data plane is
    // consumed by `pump_against` and the accepted local half speaks
    // the FRAMED wire (the N-11 contract — the pump copies raw bytes,
    // so the `[len: u16 BE]` framing lives at the boundary; the test's
    // local end stands in for the framed adapter, e.g. `local::UdpHalf`).
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let registry = ResourceRegistry::new();
    registry.register("dns", Substrate::Udp, "in-process").await;
    let topo = wire(registry, framed_udp_echo_dial()).await;

    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("dns", Substrate::Udp), None)
            .await
            .expect("reverse udp open");
    let session = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Udp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("adopt");

    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
    let session = session.pump_against(accepted_end).await;
    let mut local_end = local_end;

    // The local end speaks FRAMED bytes end-to-end: a payload frame
    // crosses the pump untouched and echoes back framed — decode
    // locally with the codec (boundary preserved, F-2 shape intact).
    let sent = alktunnels::wire::frame_datagram(b"rev-udp").expect("frame");
    local_end.write_all(&sent).await.expect("write frame");
    let mut framed_back = vec![0u8; sent.len()];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        local_end.read_exact(&mut framed_back),
    )
    .await
    .expect("round trip timed out")
    .expect("read");
    assert_eq!(framed_back, &sent[..], "framed bytes pass through");

    // An EMPTY datagram (`len=0`) also round-trips — never collapsed
    // to EOF across the pump (the F-2 invariant, reverse direction).
    local_end
        .write_all(&[0, 0])
        .await
        .expect("write empty frame");
    let mut empty_back = vec![0u8; 2];
    tokio::time::timeout(
        std::time::Duration::from_secs(5),
        local_end.read_exact(&mut empty_back),
    )
    .await
    .expect("empty round trip timed out")
    .expect("read");
    assert_eq!(empty_back, &[0, 0], "empty datagram frame survives");

    // Drop the local end: both pump directions EOF; join completes.
    drop(local_end);
    let (_, _, reaped) = session.join().await;
    assert!(reaped);
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn failed_open_never_yields_a_session() {
    // No phantom session: unknown resource → typed error, no channel
    // anywhere, and `open_ref()` exposes the typed surface.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire_forward(registry, echo_dial()).await;

    let err = match TunnelSession::open(&topo.consumer, params("ghost", Substrate::Tcp)).await {
        Ok(_) => panic!("unknown resource must not open"),
        Err(e) => e,
    };
    assert_eq!(
        err.open_ref().establishment_reason(),
        Some("unknown_resource")
    );
    assert!(only_channel_0(&topo));

    // Reverse path on the reverse topology: the call fails typed;
    // adopt is never reached.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;
    let err =
        match open_reverse_channel(&topo.consumer_call, &params("ghost", Substrate::Tcp), None)
            .await
        {
            Ok(_) => panic!("reverse open of unknown resource must fail"),
            Err(e) => e,
        };
    assert!(matches!(err, alktunnels::ReverseOpenError::Call(_)));
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn adopt_failure_is_typed_not_establishment() {
    // An adopt failure is `AdoptFailed`-class (post-reply local
    // failure: ID collision), not an establishment error. A bogus-id
    // adopt SUCCEEDS by design — the manager parks early arrivals for
    // any not-yet-seen ID (the adoption race cover, ADR-047 §5) — so
    // the collision path is the honest probe: adopt the same ID twice.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;

    let channel_id =
        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
            .await
            .expect("reverse open");

    let first = TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    .expect("first adopt");

    let err = match TunnelSession::adopt(
        &topo.consumer_manager,
        channel_id,
        Substrate::Tcp,
        alktunnels::TUNNEL_ALPN,
    )
    .await
    {
        Ok(_) => panic!("adopting a live id twice must fail"),
        Err(e) => e,
    };
    match err.open_ref() {
        alkcall::channels::client::ChannelOpenError::AdoptFailed(
            alkcall::channels::manager::ManagerError::ChannelExists(_),
        ) => {}
        other => panic!("expected AdoptFailed(ChannelExists), got {other:?}"),
    }

    first.close().await;
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
    assert!(only_channel_0(&topo));
}

#[tokio::test]
async fn concurrent_reverse_sessions_independent() {
    // Two reverse sessions on one connection: independent channels,
    // each round-trips its own marker, both reaped.
    let registry = ResourceRegistry::new();
    registry
        .register("echo", Substrate::Tcp, "in-process")
        .await;
    let topo = wire(registry, echo_dial()).await;

    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut sessions = Vec::new();
    for i in 0..2 {
        let channel_id =
            open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
                .await
                .expect("reverse open");
        let session = TunnelSession::adopt(
            &topo.consumer_manager,
            channel_id,
            Substrate::Tcp,
            alktunnels::TUNNEL_ALPN,
        )
        .await
        .expect("adopt");
        let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
        let session = session.pump_against(accepted_end).await;
        let (mut l_read, mut l_write) = tokio::io::split(local_end);
        let msg = format!("marker {i}");
        l_write.write_all(msg.as_bytes()).await.expect("write");
        l_write.flush().await.expect("flush");
        let mut buf = vec![0u8; msg.len()];
        tokio::time::timeout(
            std::time::Duration::from_secs(5),
            l_read.read_exact(&mut buf),
        )
        .await
        .expect("round trip")
        .expect("read");
        assert_eq!(buf, msg.as_bytes());
        sessions.push(session);
    }
    assert_ne!(sessions[0].channel_id, sessions[1].channel_id);

    for session in sessions {
        let (_, _, reaped) = session.join().await;
        assert!(reaped);
    }
    assert!(only_channel_0(&topo));
}

#[allow(dead_code)]
fn type_assertions() {
    // The `compile_fail` doc-test in consumer.rs is the no-Clone gate;
    // this parity stub keeps the type in the test's namespace.
    let _ = std::mem::size_of::<TunnelSession>();
}