//! The `w3-tunnel-direct` contract rows: the tunnel branch served for
//! real: the route-1 bridge (ADR-006 decision 1) lands
//! `direct-tcpip`/`direct-streamlocal` through the published alktunnels
//! 0.5.0 seam (alkssh 0.2.0's W2 adoption: `TunnelSession::open_direct`
//! → accept only after the fabric open succeeds → `take_halves()`
//! pumped against the taken channel; the forwarded serving surface on
//! the establisher shape, the `ForwardedGate` correlation gate),
//! observed through the loopback harness against the **fabric
//! double** (`w3-tunnel-direct`'s recorded test composition: the
//! POC-3 §C.4 in-process recipe (`ChannelsAdapter` +
//! `InstallChannelZero` + the alktunnels openables + a hand-rolled
//! dial/dial-fail seam, hand-rolled here, never a sibling-checkout
//! path).
//!
//! The decline-matrix classes *per feature build* stay in
//! `tests/dispatch_core.rs` (unfeatured rows in the featureless/git/
//! session builds; the absent-handle rows in the tunnel builds); this
//! binary adds the featured-build rows the fabric double makes
//! observable: the serve class (data round trip, both substrates), the
//! fabric-failure class (`ConnectFailed`), the cap classes (refusal +
//! the release mechanism: an admitted-but-rejected bridge releases its
//! count, a closed serving channel recovers), the composed identity
//! gate (the fabric's `AccessControl::check` refusing an unscoped
//! channels identity, the visible-surface invariant's alkcall half,
//! with the establisher never reached), the establisher's identity
//! witness (the POC-3 §C.3 shape), the OQ-11 off-Unix gate, and the
//! real-tool `ssh -L` smoke.
//!
//! The fabric double's dial seam is in-process by design: the dial
//! target is matched against a map of pre-created duplex halves keyed
//! by the rendered target string (`host:port` / socket path, exactly
//! the shape the establisher hands the `DialFn`), so the serve rows
//! are fully deterministic (no sockets on the target substrate), while
//! the real-tool smoke hand-rolls a real-TCP dial closure against a
//! real loopback endpoint (the smoke exercises the substrate for real;
//! tooling posture recorded like the git task, skipped-with-reason
//! when `ssh`/`curl` is unavailable, never silently green).

#![allow(dead_code)]

mod common;

#[cfg(feature = "tunnel")]
mod serving {
    use std::collections::HashMap;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};

    use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
    use alkcall::channels::client::{ChannelClient, ServingConfig};
    use alkcall::channels::operations::{ChannelCore, ChannelOperations, Establishment};
    use alkcall::core::auth::{AuthContext, Identity, NoopIdentityProvider};
    use alkcall::core::types::{Connection as CoreConnection, ProtocolHandler};
    use alkcall::protocol::connection::{CallConnection, split_single_stream};
    use alkcall::protocol::dispatch::Dispatcher;
    use alkcall::registry::registration::OperationRegistry;
    use alktunnels::params::{
        Substrate, SubstrateAddr, TUNNEL_DIRECT_SCOPE, TUNNEL_FORWARDED_SCOPE, TUNNEL_OPEN_SCOPE,
    };
    use alktunnels::producer::{
        BindFirstAcceptLoop, BindFirstPlan, BindFn, BoundListener, DialFn, ForwardedOpener,
        IdentityWitness, TargetHandle, TunnelEstablishError, make_bindfirst_accept_handler,
        register_tunnel_direct_openable,
    };
    use russh::client::Msg;
    use russh::{Channel, ChannelMsg, ChannelOpenFailure, Error};
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::sync::mpsc;

    use super::common::harness::{
        ForwardedOpen, LoopbackSession, cancel_tcpip_forward, drive, open_direct_streamlocal,
        open_direct_tcpip, serve_config as harness_serve_config, tcpip_forward,
    };
    use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
    use alkssh::{Limits, SshAdapter, TunnelListenSeam};

    type TargetMap = Arc<std::sync::Mutex<HashMap<String, tokio::io::DuplexStream>>>;

    /// The in-process fabric (`w3-tunnel-direct`'s recorded double, the
    /// POC-3 §C.4 recipe): an `alk/channels` connection over a tokio
    /// duplex whose accept side registers `ChannelOperations` + the
    /// `channels/tunnel/direct` openable (the establisher's dial seam
    /// injected), and whose connect side is the `Arc<ChannelClient>` a
    /// tunnel-featured serve-config holds. The accept side's channel-0
    /// identity (`caller`) is what the fabric's
    /// `AccessControl::check(peer_identity)` gates the mapped opens
    /// against, the assembly-composed identity of the real deployment;
    /// `witness` records what the establisher's per-call auth (alkcall
    /// CF-006) saw.
    pub(super) async fn wire_fabric(
        dial: DialFn,
        caller: Identity,
        witness: IdentityWitness,
    ) -> Arc<ChannelClient> {
        let registry = Arc::new(OperationRegistry::new());
        let registry_for_hook = Arc::clone(&registry);
        let witness_for_hook = Arc::clone(&witness);
        let caller_for_hook = caller.clone();
        let dial_for_hook = Arc::clone(&dial);

        let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
            let registry = Arc::clone(&registry_for_hook);
            let witness = Arc::clone(&witness_for_hook);
            let caller = caller_for_hook.clone();
            let dial = Arc::clone(&dial_for_hook);
            tokio::spawn(async move {
                let Ok(channel0_bidi) = channel0_conn.accept_bi().await else {
                    return;
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                // The serving-side identity set once at install, the
                // hub_leg.rs one-time-set precedent the door's
                // `set_identity` mechanics decision reads (the identity
                // the serving dispatch resolves for the mapped opens'
                // ACL).
                let _ = channel0_conn.set_identity(caller);
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));
                let ops = ChannelOperations::with_default_policy(manager.clone());
                ops.register_on(&registry).expect("channel ops register");
                register_tunnel_direct_openable(
                    &ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()),
                    &registry,
                    AuthContext::anonymous(b"alk/tunnel"),
                    dial,
                    Some(witness),
                )
                .expect("tunnel direct openable registers");
                let dp = Dispatcher::new(registry, Arc::new(NoopIdentityProvider));
                dp.serve_single_stream(call_connection, reader, writer)
                    .await;
            })
        });

        let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
        let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
        let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);

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

        Arc::new(
            ChannelClient::from_connection(consumer_conn)
                .await
                .expect("the fabric's channels client establishes"),
        )
    }

    /// The leg-2 fabric double (`w3-tunnel-listen`'s recorded
    /// composition): the leg-1 wire fabric extended with the
    /// **bind-first listen half**, the assembly-composed shape the
    /// vanilla `-R` listen flow rides, built exactly on the published
    /// alktunnels pieces: a `bindfirst_establisher`-shaped
    /// establisher, `make_bindfirst_accept_handler`, and the
    /// `ForwardedOpener` dialing the forwarded op back at the door.
    ///
    /// The **door-side serving surface**: the `Arc<ChannelClient>`
    /// constructed with `from_connection_with_serving` against a
    /// shared registry the row hands alkssh through the
    /// `TunnelListenSeam` (one `Arc`, both sides).
    ///
    /// The double's bind seam is the **dynamic-address producer** the
    /// door's naming convention front: the establisher treats the
    /// wire-mapped resource string as the bind endpoint directly
    /// (`resource = "127.0.0.1:9200"` binds `127.0.0.1:9200`) and mints
    /// the correlation token per the ADR-008 discipline. Who may bind
    /// what is the fabric's policy: the double binds what the room
    /// asks; a real fabric's ACL is the same seam. `localhost`
    /// renders to `127.0.0.1` so the real-tool smoke's curl is
    /// deterministic about the family.
    pub(super) struct FabricDouble {
        pub(super) client: Arc<ChannelClient>,
        pub(super) registry: Arc<OperationRegistry>,
        opener: Arc<tokio::sync::OnceCell<ForwardedOpener>>,
        /// The establisher's mint stream (the rows' real-token vehicle
        /// which the ACL row needs; the genuinely minted token its listen's
        /// fabric established; the SSH layer never carries the token).
        minted: tokio::sync::Mutex<mpsc::UnboundedReceiver<u32>>,
    }

    impl FabricDouble {
        /// The fabric's `ForwardedOpener` (the call plane), exported
        /// for the rows' direct drives (an arrival the fabric would
        /// make, dialed by the row itself (the unknown-token and
        /// composed-gate rows' vehicle).
        pub(super) async fn forwarded_opener(&self) -> ForwardedOpener {
            for _ in 0..600 {
                if let Some(opener) = self.opener.get() {
                    return Arc::clone(opener);
                }
                tokio::task::yield_now().await;
            }
            panic!("the fabric's forwarded opener never installed");
        }

        /// Drives the fabric's forwarded-open closure directly:
        /// `Ok` carries the adopted fabric-side halves (the door
        /// allocated the channel, so the gate-pass/residual-race
        /// teardown-at-birth is then observable as an EOF on the
        /// `recv` half), `Err` is the door-side refusal (the
        /// unknown-token gate miss / the composed-gate row's ACL
        /// answer; the arrival never became anything).
        /// The next minted correlation token (one per accepted listen,
        /// in mint order (the rows' real-token vehicle).
        pub(super) async fn minted_token(&self) -> u32 {
            self.minted
                .lock()
                .await
                .recv()
                .await
                .expect("the fabric's establisher minted a token in time")
        }

        pub(super) async fn open_forwarded_directly(
            &self,
            params: alktunnels::params::TunnelForwardedParams,
        ) -> Result<
            (
                alkcall::channels::reassembly::MpscSendStream,
                alkcall::channels::reassembly::MpscRecvStream,
            ),
            TunnelEstablishError,
        > {
            let opener = self.forwarded_opener().await;
            opener(params).await
        }
    }

    /// The double's establisher mint discipline: `Counter` is the
    /// ADR-008 shape the real fabric carries (a non-zero-seeded
    /// counter, never 0, never re-minted); `WithCollisions` is the
    /// row-pinned collision construct; the first `collisions` mints
    /// return a fixed `token` (the test forces what the fabric's
    /// discipline prevents), then the counter continues from where it
    /// left.
    #[derive(Clone, Copy)]
    pub(super) enum MintStrategy {
        Counter,
        WithCollisions { token: u32, collisions: usize },
    }

    /// The dynamic-bind producer's establisher (the double's shape of
    /// the `bindfirst_establisher` recipe): parse
    /// `TunnelParams {resource, substrate}`, bind via the injected
    /// [`BindFn`] directly on the resource string, mint the token per
    /// the ADR-008 discipline (non-zero-seeded counter, never 0), and
    /// contribute the `bound` + `listen` reply fields (ADR-049
    /// amendment 3).
    fn dynamic_bindfirst_establisher(
        bind: BindFn,
        minted: Arc<mpsc::UnboundedSender<u32>>,
        mint: MintStrategy,
    ) -> alkcall::channels::operations::OpenEstablisher {
        let counter = Arc::new(AtomicU32::new(1));
        let (collided_token, remaining_collisions) = match mint {
            MintStrategy::Counter => (0, 0usize),
            MintStrategy::WithCollisions { token, collisions } => (token, collisions),
        };
        let collisions = Arc::new(AtomicUsize::new(remaining_collisions));
        Arc::new(move |input: serde_json::Value, _auth| {
            let bind = Arc::clone(&bind);
            let minted = Arc::clone(&minted);
            let counter = Arc::clone(&counter);
            let collisions = Arc::clone(&collisions);
            Box::pin(async move {
                let params: alktunnels::params::TunnelParams =
                    serde_json::from_value(input.clone()).map_err(|e| {
                        TunnelEstablishError::HandlerError(format!("invalid params: {e}"))
                    })?;
                let token = if collided_token != 0 && collisions.load(Ordering::Relaxed) > 0 {
                    collisions.fetch_sub(1, Ordering::Relaxed);
                    collided_token
                } else {
                    let next = counter.fetch_add(1, Ordering::Relaxed);
                    if next == 0 { 1 } else { next }
                };
                let _ = minted.send(token);
                let BoundListener { listener, bound } =
                    bind(params.substrate, &params.resource).await?;
                let plan = BindFirstPlan {
                    listener,
                    token,
                    substrate: params.substrate,
                };
                Ok(Establishment::new(Arc::new(plan))
                    .with_reply_field(
                        "bound",
                        serde_json::to_value(&bound)
                            .map_err(|e| TunnelEstablishError::HandlerError(format!("{e}")))?,
                    )
                    .with_reply_field("listen", serde_json::json!(token)))
            })
        })
    }

    /// The double's real-socket bind seam: real TCP/unix binds of the
    /// wire-named endpoint (the platform boundary the
    /// `tunnels/local` feature serves in a real deployment,
    /// hand-rolled here, never a sibling-checkout path). Unix binds
    /// unlink a stale socket file first (`bind_unix`'s caller
    /// posture).
    fn real_bind() -> BindFn {
        Arc::new(move |substrate, endpoint: &str| {
            let endpoint = endpoint.to_owned();
            Box::pin(async move {
                match substrate {
                    alktunnels::params::Substrate::Unix => {
                        let _ = std::fs::remove_file(&endpoint);
                        let listener = tokio::net::UnixListener::bind(&endpoint).map_err(|e| {
                            TunnelEstablishError::ResourceShortage(format!(
                                "bind `{endpoint}`: {e}"
                            ))
                        })?;
                        Ok(BoundListener {
                            listener: Arc::new(listener),
                            bound: SubstrateAddr::Path {
                                path: endpoint.clone(),
                            },
                        })
                    }
                    _ => {
                        let hostport = endpoint.replace("localhost:", "127.0.0.1:");
                        let listener =
                            tokio::net::TcpListener::bind(&hostport)
                                .await
                                .map_err(|e| {
                                    TunnelEstablishError::ResourceShortage(format!(
                                        "bind `{hostport}`: {e}"
                                    ))
                                })?;
                        let addr = listener.local_addr().expect("bound tcp addr");
                        Ok(BoundListener {
                            listener: Arc::new(listener),
                            bound: SubstrateAddr::HostPort {
                                host: addr.ip().to_string(),
                                port: addr.port(),
                            },
                        })
                    }
                }
            })
        })
    }

    /// The double's accept loop: real TCP/unix accept of ONE pair per
    /// invocation (the dispatcher's per-accept contract), downcasting
    /// the plan's opaque listener to the bind seam's own wrappers. The
    /// unix leg's `peer` is fabricated (an unnamed peer cannot ride
    /// the forwarded op's required `peer` param (the shipped
    /// accept-pair posture).
    fn real_accept_loop() -> BindFirstAcceptLoop {
        Arc::new(move |listener: Arc<dyn std::any::Any + Send + Sync>| {
            Box::pin(async move {
                if let Some(tcp) = listener.downcast_ref::<tokio::net::TcpListener>() {
                    let (stream, peer) = tcp
                        .accept()
                        .await
                        .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?;
                    let (read, write) = stream.into_split();
                    return Ok((
                        SubstrateAddr::HostPort {
                            host: peer.ip().to_string(),
                            port: peer.port(),
                        },
                        TargetHandle {
                            read: Box::new(read),
                            write: Box::new(write),
                        },
                    ));
                }
                if let Some(unix) = listener.downcast_ref::<tokio::net::UnixListener>() {
                    let (stream, _peer) = unix
                        .accept()
                        .await
                        .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?;
                    let (read, write) = stream.into_split();
                    return Ok((
                        SubstrateAddr::Path {
                            path: "/fabric-arrival-peer.sock".to_owned(),
                        },
                        TargetHandle {
                            read: Box::new(read),
                            write: Box::new(write),
                        },
                    ));
                }
                Err(TunnelEstablishError::HandlerError(
                    "the plan's listener is not one of the double's binds".into(),
                ))
            })
        })
    }

    /// Wires the leg-2 fabric double with the ADR-008 mint discipline
    /// (the rows' default mint).
    pub(super) async fn wire_fabric_with_listen(
        door_scoped_caller: bool,
        arrival_scoped: bool,
    ) -> FabricDouble {
        wire_fabric_with_listen_mint(door_scoped_caller, arrival_scoped, MintStrategy::Counter)
            .await
    }

    /// Wires the leg-2 fabric double (leg 1 unchanged underneath, the
    /// dial seam + direct openable stay exactly `wire_fabric`'s
    /// composition, against a leg-1 fail dial the -R rows never
    /// exercise). `door_scoped_caller` composes the fabric-side
    /// serving identity (what gates the door's listen op's ACL there);
    /// `arrival_scoped` composes the door-side `ServingConfig.identity`
    /// with or without the `tunnel:forwarded` scope (the composed-gate
    /// row's negative leg). `mint` is the establisher's mint
    /// discipline ([`MintStrategy`]), the collision rows' vehicle.
    pub(super) async fn wire_fabric_with_listen_mint(
        door_scoped_caller: bool,
        arrival_scoped: bool,
        mint: MintStrategy,
    ) -> FabricDouble {
        // The two sides' registries are independent (the fabric's own
        // dispatch registry; and the DOOR-side shared one the row
        // hands alkssh's seam; the registration into the fabric's
        // registry would be the silent-fork shape). The earlier draft
        // wired one Arc to both, pinned wrong here for exactly one
        // reason: the door's `tunnel/forwarded` openable must live in
        // the DOOR's dispatch surface alone.
        let door_registry = Arc::new(OperationRegistry::new());
        let registry = Arc::new(OperationRegistry::new());
        let registry_for_hook = Arc::clone(&registry);
        let door_caller = if door_scoped_caller {
            identity("door-caller", &[TUNNEL_OPEN_SCOPE])
        } else {
            identity("door-caller-no-scope", &["ssh"])
        };
        let arrival_identity = if arrival_scoped {
            identity("fabric-arrival", &[TUNNEL_FORWARDED_SCOPE])
        } else {
            identity("fabric-arrival-plain", &["ssh"])
        };
        let dial: DialFn = Arc::new(
            move |_substrate: alktunnels::params::Substrate, target: &str| {
                let target = target.to_owned();
                Box::pin(async move {
                    Err::<TargetHandle, TunnelEstablishError>(TunnelEstablishError::DialFailed(
                        format!("the leg-1 dial double is not part of the -R rows: {target}"),
                    ))
                })
            },
        );
        let opener_cell = Arc::new(tokio::sync::OnceCell::new());
        let opener_for_hook = Arc::clone(&opener_cell);
        let (minted_sender, minted_receiver) = mpsc::unbounded_channel::<u32>();
        let minted_tx: Arc<mpsc::UnboundedSender<u32>> = Arc::new(minted_sender);
        let minted_for_hook = minted_tx.clone();

        let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
            let registry = Arc::clone(&registry_for_hook);
            let door_caller = door_caller.clone();
            let minted_tx = minted_for_hook.clone();
            let dial = Arc::clone(&dial);
            let opener_cell = Arc::clone(&opener_for_hook);
            tokio::spawn(async move {
                let Ok(channel0_bidi) = channel0_conn.accept_bi().await else {
                    return;
                };
                let (writer, reader) = split_single_stream(channel0_bidi);
                let _ = channel0_conn.set_identity(door_caller);
                let call_connection = Arc::new(CallConnection::new_single_stream(
                    channel0_conn,
                    Arc::clone(&writer),
                ));

                // The fabric's ForwardedOpener (the call plane):
                // dial the forwarded op at the door, adopt the
                // door-allocated id on the fabric's manager, and
                // return the adopted channel's halves
                // (make_bindfirst_accept_handler composes the
                // pump). Exported for the rows' direct drives.
                let open_forwarded: ForwardedOpener = {
                    let call = Arc::clone(&call_connection);
                    let manager = manager.clone();
                    Arc::new(move |params: alktunnels::params::TunnelForwardedParams| {
                        let call = Arc::clone(&call);
                        let manager = manager.clone();
                        Box::pin(async move {
                            let payload = serde_json::to_value(&params)
                                .map_err(|e| TunnelEstablishError::HandlerError(format!("{e}")))?;
                            let response = call
                                .call(alktunnels::params::OP_TUNNEL_FORWARDED, payload)
                                .await;
                            // The real opener maps the wire refusal's
                            // reason through the `From<TunnelEstablish
                            // Error> for EstablishmentError` table; the
                            // typed `unknown_resource` (the gate's miss
                            // class) must survive the client side here,
                            // so the double maps the details.reason back
                            // into the establisher's taxonomy (the flat
                            // `HandlerError` swallow would erase exactly
                            // what the gate rows pin).
                            let out = response.result.map_err(|e| {
                                if e.code == "channel:open_failed" {
                                    match e
                                        .details
                                        .as_ref()
                                        .and_then(|d| d.get("reason"))
                                        .and_then(|r| r.as_str())
                                        .unwrap_or_default()
                                    {
                                        "unknown_resource" => {
                                            TunnelEstablishError::UnknownResource(e.message.clone())
                                        }
                                        "dial_failed" => {
                                            TunnelEstablishError::DialFailed(e.message.clone())
                                        }
                                        "resource_shortage" => {
                                            TunnelEstablishError::ResourceShortage(
                                                e.message.clone(),
                                            )
                                        }
                                        "handler_error" => {
                                            TunnelEstablishError::HandlerError(e.message.clone())
                                        }
                                        _ => TunnelEstablishError::HandlerError(format!(
                                            "forwarded open: {e:?}"
                                        )),
                                    }
                                } else {
                                    TunnelEstablishError::HandlerError(format!(
                                        "forwarded open: {e:?}"
                                    ))
                                }
                            })?;
                            let id =
                                out.get("channel_id")
                                    .and_then(|v| v.as_u64())
                                    .ok_or_else(|| {
                                        TunnelEstablishError::HandlerError("no channel_id".into())
                                    })? as u32;
                            manager
                                .adopt_channel(id, alktunnels::params::TUNNEL_ALPN, None)
                                .await
                                .map_err(|e| {
                                    TunnelEstablishError::HandlerError(format!("adopt: {e:?}"))
                                })
                        })
                    })
                };
                let _ = opener_cell.set(Arc::clone(&open_forwarded));

                let ops = ChannelOperations::with_default_policy(manager.clone());
                ops.register_on(&registry).expect("channel ops register");
                register_tunnel_direct_openable(
                    &ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy()),
                    &registry,
                    AuthContext::anonymous(b"alk/tunnel"),
                    dial,
                    None,
                )
                .expect("the direct openable registers");
                let core =
                    ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy());
                core.register_openable_with_establisher(
                    alktunnels::params::tunnel_open_spec(),
                    Some(dynamic_bindfirst_establisher(
                        real_bind(),
                        Arc::clone(&minted_tx),
                        mint,
                    )),
                    make_bindfirst_accept_handler(real_accept_loop(), open_forwarded),
                    &registry,
                    AuthContext::anonymous(b"alk/tunnel"),
                    None,
                )
                .expect("the bind-first listen openable registers");

                let dp = Dispatcher::new(registry, Arc::new(NoopIdentityProvider));
                dp.serve_single_stream(call_connection, reader, writer)
                    .await;
            })
        });

        let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
        let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
        let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);

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

        let client = ChannelClient::from_connection_with_serving(
            consumer_conn,
            Some(ServingConfig {
                registry: Arc::clone(&door_registry),
                identity_provider: Arc::new(NoopIdentityProvider),
                identity: Some(arrival_identity),
            }),
        )
        .await
        .expect("the fabric's channels client establishes");

        FabricDouble {
            client: Arc::new(client),
            registry: door_registry,
            opener: opener_cell,
            minted: tokio::sync::Mutex::new(minted_receiver),
        }
    }

    /// A pre-created duplex target, the "in-process dial" half: the
    /// test holds the returned side, the map hands the dial seam the
    /// other half when the establisher renders a matching target.
    fn inproc_target(targets: &TargetMap, key: &str, capacity: usize) -> tokio::io::DuplexStream {
        let (target_side, test_side) = tokio::io::duplex(capacity);
        targets
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .insert(key.to_owned(), target_side);
        test_side
    }

    /// The hand-rolled dial seam (the assembly's injected `DialFn`,
    /// doubled): an exact `host:port` / socket-path key pops the
    /// pre-created half; an absent key is the dial-fail seam
    /// (`TunnelEstablishError::DialFailed`); both the fabric-failure
    /// rows' shape and the release rows' failure leg. The call counter
    /// witnesses the "no fabric call" assertions.
    fn inproc_dial(targets: TargetMap, calls: Arc<AtomicUsize>) -> DialFn {
        Arc::new(move |_substrate: Substrate, target: &str| {
            let targets = Arc::clone(&targets);
            let calls = Arc::clone(&calls);
            let target = target.to_owned();
            Box::pin(async move {
                calls.fetch_add(1, Ordering::SeqCst);
                let half = targets
                    .lock()
                    .unwrap_or_else(|e| e.into_inner())
                    .remove(&target)
                    .ok_or_else(|| {
                        TunnelEstablishError::DialFailed(format!(
                            "in-process double target `{target}` absent"
                        ))
                    })?;
                let (read, write) = tokio::io::split(half);
                Ok(TargetHandle {
                    read: Box::new(read),
                    write: Box::new(write),
                })
            })
        })
    }

    /// The fabric caller identity with the direct op's scope, the
    /// composed-gate identity the serving side resolves, so the ACL
    /// admits the mapped opens.
    fn scoped_caller() -> Identity {
        identity("tunnel-caller", &[TUNNEL_DIRECT_SCOPE])
    }

    /// The tunnel build's authenticated loopback session over the
    /// doubles, with the fabric client wired in: the fixture identity
    /// resolves through the provider and the serve-config carries the
    /// `channel_client` (the route-1 bridge's family-fabric entry);
    /// `limits` rides the serve-config (the cap rows tune them).
    async fn tunnel_session(client: Arc<ChannelClient>, limits: Limits) -> LoopbackSession {
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], limits);
        serve.channel_client = Some(client);
        let adapter = SshAdapter::new(serve)
            .await
            .expect("the tunnel build's adapter constructs");

        let policy = Arc::new(MemHostKeyPolicy::with_default(true));
        let mut session = drive(adapter, Arc::clone(&policy), Limits::default())
            .await
            .expect("the loopback mounts");

        let result =
            authenticate_publickey(&mut session.client, "tunnel-rider", fixture("alpha")).await;
        assert!(
            result.success(),
            "the fixture identity authenticates before every tunnel row"
        );
        session
    }

    /// The harness import that needs the loopback session's client
    /// (kept near the top of the module for the rows' readability).
    use super::common::harness::authenticate_publickey;

    /// Matches an open failure's SSH reason, the wire decline reason
    /// the rows assert.
    fn open_failure(err: Error, expected: ChannelOpenFailure, context: &str) {
        assert!(
            matches!(&err, Error::ChannelOpenFailure(reason) if *reason == expected),
            "{context}: expected open decline with {expected:?}, got {err:?}"
        );
    }

    /// The serve row's data leg: the SSH channel against the in-process
    /// target: one request down `direct-tcpip`, one reply back (the
    /// bidirectional pump of the route-1 bridge, the POC-3 §C.1 shape:
    /// zero framing, both directions live).
    async fn round_trip(
        channel: &mut Channel<Msg>,
        mut target_side: tokio::io::DuplexStream,
        request: &[u8],
        reply: &[u8],
        context: &str,
    ) {
        channel
            .data_bytes(request.to_vec())
            .await
            .expect("the SSH channel carries the request down");

        let mut seen_request = vec![0u8; request.len()];
        tokio::time::timeout(
            std::time::Duration::from_secs(15),
            target_side.read_exact(&mut seen_request),
        )
        .await
        .expect("the request arrives through the bridge in time")
        .expect("the pump delivers the request to the substrate");
        assert_eq!(
            seen_request, request,
            "{context}: the bridge pumps the SSH channel bytes unframed"
        );

        target_side
            .write_all(reply)
            .await
            .expect("the target half writes the reply");
        target_side.flush().await.expect("the flush rides");

        let mut seen_reply = Vec::new();
        tokio::time::timeout(std::time::Duration::from_secs(15), async {
            while seen_reply.len() < reply.len() {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen_reply.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("{context}: the channel ended before the reply arrived"),
                }
            }
        })
        .await
        .expect("the reply arrives through the bridge in time");
        seen_reply.truncate(reply.len());
        assert_eq!(
            seen_reply, reply,
            "{context}: the bridge pumps the substrate bytes back unframed"
        );
    }

    /// - *Contract stamp:* ADR-006 decision 1 (the route-1 bridge: the
    ///   `direct-tcpip` open → `TunnelSession::open_direct` over a held
    ///   `ChannelClient` → accept only after the fabric open succeeds →
    ///   `take_halves()` pumped bidirectionally (the POC-3 §C.1 shape,
    ///   zero framing both directions; the open hook returns fast).
    #[tokio::test]
    async fn direct_tcpip_serves_through_the_route_1_bridge() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let witness = Arc::new(tokio::sync::Mutex::new(None));
        let target_side = inproc_target(&targets, "127.0.0.1:36001", 4096);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), Arc::clone(&calls)),
            scoped_caller(),
            Arc::clone(&witness),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36001)
            .await
            .expect("the featured direct-tcpip row serves through the bridge");
        round_trip(
            &mut channel,
            target_side,
            b"GET /hello HTTP/1.1\r\nHost: target.example\r\n\r\n",
            b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello",
            "the direct-tcpip serve row",
        )
        .await;

        let witnessed = witness.lock().await.clone();
        assert_eq!(
            witnessed.as_deref(),
            Some("tunnel-caller"),
            "the direct establisher's per-call auth saw the channels connection's identity (CF-006, POC-3 §C.3's witness)"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "one fabric dial rode the open"
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows
    ///   (`direct-streamlocal` → `SubstrateAddr::Path` +
    ///   `Substrate::Unix`, the route-1 bridge's second substrate;
    ///   ADR-006 decision 1).
    #[cfg(unix)]
    #[tokio::test]
    async fn direct_streamlocal_serves_through_the_route_1_bridge_on_unix() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let target_side = inproc_target(&targets, "/inproc/tunnel-target.sock", 4096);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), calls),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let mut channel = open_direct_streamlocal(&session.client, "/inproc/tunnel-target.sock")
            .await
            .expect("the on-Unix direct-streamlocal row serves through the bridge");
        round_trip(
            &mut channel,
            target_side,
            b"unix-request-bytes",
            b"unix",
            "the direct-streamlocal serve row",
        )
        .await;

        session.unmount().await;
    }

    /// - *Contract stamp:* ADR-006 decision 1 (fabric-open failure →
    ///   the explicit `ConnectFailed` reject (accept only after the
    ///   fabric open succeeds; the wave-2 stub's reason surviving as
    ///   the real decline; POC-3 §C.2's failure shape).
    #[tokio::test]
    async fn a_failed_fabric_open_rejects_the_open_connect_failed() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let client = wire_fabric(
            inproc_dial(targets, Arc::clone(&calls)),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let err = open_direct_tcpip(&session.client, "unreachable.example", 9)
            .await
            .expect_err("the fabric open fails on the absent dial target");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the fabric-failure direct-tcpip row",
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            1,
            "the bridge dialed the fabric exactly once; the SSH open carried the fabric's answer"
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* `w3-tunnel-direct`'s recorded cap release
    ///   mechanism (the admission is taken at the hook and held across
    ///   the fabric open; a fabric failure releases the count **inside
    ///   the spawned task**; the `channel_close` release point only
    ///   fires for established channels, so without this release a
    ///   failed bridge would leak budget): a failed bridge under a
    ///   full cap does not strand the budget; the next open of the
    ///   same session is admitted (it would answer
    ///   `AdministrativelyProhibited` if the count had leaked).
    #[tokio::test]
    async fn a_failed_bridge_releases_the_cap_count_before_the_next_open() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let target_side = inproc_target(&targets, "127.0.0.1:36002", 512);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), calls),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let limits = Limits {
            max_concurrent_channels: 1,
            ..Limits::default()
        };
        let session = tunnel_session(client, limits).await;

        let err = open_direct_tcpip(&session.client, "unreachable.example", 9)
            .await
            .expect_err("the failing bridge declines");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the cap-release row's failing leg",
        );

        // The failure answer landed before the client saw it
        // (release-then-reject inside the task), so the very next open
        // of this same session observes the released count.
        let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36002)
            .await
            .expect("the next open is admitted; the failed bridge released its count");
        round_trip(
            &mut channel,
            target_side,
            b"req",
            b"rep",
            "the cap-release row's serving leg",
        )
        .await;

        session.unmount().await;
    }

    /// - *Contract stamp:* AGENTS.md convention 9 / `overview.md`
    ///   §Bounded-resources policy + `ChannelCap`'s recorded
    ///   disjoint-release shape: a *closed serving* channel releases
    ///   through the `channel_close` hook; a full cap refuses with the
    ///   cap reason, and a recovered cap admits again.
    #[tokio::test]
    async fn the_cap_refuses_beyond_the_ceiling_and_recovers_on_close() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let target_side = inproc_target(&targets, "127.0.0.1:36003", 512);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), calls),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let limits = Limits {
            max_concurrent_channels: 1,
            ..Limits::default()
        };
        let session = tunnel_session(client, limits).await;

        let mut channel = open_direct_tcpip(&session.client, "127.0.0.1", 36003)
            .await
            .expect("the first open admits under the cap");
        round_trip(
            &mut channel,
            target_side,
            b"q",
            b"a",
            "the cap-refusal row's serving channel",
        )
        .await;

        let err = open_direct_tcpip(&session.client, "127.0.0.1", 36003)
            .await
            .expect_err("the second open rides beyond the cap");
        open_failure(
            err,
            ChannelOpenFailure::AdministrativelyProhibited,
            "the cap-exceeded direct-tcpip row",
        );

        channel.close().await.expect("the serving channel closes");
        // The close cycles back through the packet loop, the
        // established-channel release fires, and the cap recovers. The
        // recovery is observable because the dial target is consumed
        // once: the moment the cap admits again, the open proceeds
        // onto the bridge (and fails at the fabric dial with the
        // connect-failed class); an `AdministrativelyProhibited`
        // answer here would mean a leaked count.
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            let mut attempts = 0u32;
            loop {
                match open_direct_tcpip(&session.client, "127.0.0.1", 36003).await {
                    Ok(opened) => {
                        drop(opened);
                        return;
                    }
                    Err(Error::ChannelOpenFailure(
                        ChannelOpenFailure::AdministrativelyProhibited,
                    )) if attempts < 40 => {
                        attempts += 1;
                        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                    }
                    Err(Error::ChannelOpenFailure(
                        ChannelOpenFailure::AdministrativelyProhibited,
                    )) => {
                        panic!("the cap never recovered after the close; the count leaked")
                    }
                    Err(Error::ChannelOpenFailure(_)) => {
                        // The open went past the cap answer and onto the
                        // bridge (the fabric dial's connect-failed); the
                        // released count is what this row observes.
                        return;
                    }
                    Err(other) => panic!("unexpected open error during recovery: {other:?}"),
                }
            }
        })
        .await
        .expect("the cap recovery row resolves in time");

        session.unmount().await;
    }

    /// - *Contract stamp:* AGENTS.md convention 7 (the visible-surface
    ///   invariant's alkcall half): the fabric's
    ///   `AccessControl::check(peer_identity)` gates the mapped open:
    ///   a channels connection whose identity lacks the direct op's
    ///   scope is refused by the fabric's dispatch (`FORBIDDEN`), the
    ///   establisher never runs, and the door answers the SSH open
    ///   with the explicit `ConnectFailed`. The auth bridge's
    ///   SSH-resolved identity and the fabric leg's dial identity
    ///   compose at the assembly (the deployment policy); this row is
    ///   the composed-gate witness through the double.
    #[tokio::test]
    async fn the_fabric_acl_refuses_an_unscoped_channels_identity() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let witness = Arc::new(tokio::sync::Mutex::new(None));
        let _ = inproc_target(&targets, "127.0.0.1:36004", 512);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), Arc::clone(&calls)),
            identity("tunnel-caller-no-scope", &["ssh"]),
            Arc::clone(&witness),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let err = open_direct_tcpip(&session.client, "127.0.0.1", 36004)
            .await
            .expect_err("the unscoped channels identity is gated at the fabric");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the fabric-ACL direct-tcpip row",
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "the ACL check precedes the establisher; a refused open never dialed"
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (`direct-tcpip`'s
    ///   target is the SSH open arguments verbatim; an impossible wire
    ///   target, a port outside the u16 space, is a declined
    ///   pre-admission, `invalid-target`, connect-failed on the wire).
    #[tokio::test]
    async fn an_out_of_range_wire_port_declines_pre_admission() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let client = wire_fabric(
            inproc_dial(targets, Arc::clone(&calls)),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let err = open_direct_tcpip(&session.client, "target.example", u32::from(u16::MAX) + 1)
            .await
            .expect_err("an out-of-range wire port declines");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the invalid-target direct-tcpip row",
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "decline-first, alloc never: an impossible target never reaches the fabric"
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* OQ-11 (the runtime-Unix gate: universal
    ///   compile, clean open-decline off-Unix **before any fabric
    ///   call**, never a partial bridge): this row compiles in this
    ///   binary only off-Unix and pins the gate with its dial-call
    ///   witness at zero. On-Unix builds ride the serve row above.
    #[cfg(not(unix))]
    #[tokio::test]
    async fn direct_streamlocal_declines_off_unix_before_any_fabric_call() {
        let targets: TargetMap = Arc::default();
        let calls = Arc::new(AtomicUsize::new(0));
        let _ = inproc_target(&targets, "/inproc/off-unix.sock", 512);
        let client = wire_fabric(
            inproc_dial(Arc::clone(&targets), Arc::clone(&calls)),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let session = tunnel_session(client, Limits::default()).await;

        let err = open_direct_streamlocal(&session.client, "/inproc/off-unix.sock")
            .await
            .expect_err("the off-Unix streamlocal open declines");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the OQ-11 off-Unix direct-streamlocal row",
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            0,
            "the runtime gate precedes any fabric call; no partial bridge"
        );

        session.unmount().await;
    }

    // --- the `-R` listen rows (w3-tunnel-listen) ---------------------------

    /// The tunnel build's authenticated loopback session over the
    /// leg-2 double, with the fabric leg + listen seam wired (the
    /// `-R` bridge's serve-config halves) and the harness client's
    /// acceptor installed (the matching-`-R`-forward stand-in). The
    /// returned receiver is the forwarded-opens observation channel.
    async fn listen_session(
        double: &FabricDouble,
        limits: Limits,
    ) -> (
        LoopbackSession,
        tokio::sync::mpsc::UnboundedReceiver<ForwardedOpen>,
    ) {
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], limits);
        serve.channel_client = Some(Arc::clone(&double.client));
        serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let adapter = SshAdapter::new(serve)
            .await
            .expect("the -R wired adapter constructs");

        let policy = Arc::new(MemHostKeyPolicy::with_default(true));
        let (acceptor, receiver) = super::common::harness::ForwardedAcceptor::new();
        let mut session = super::common::harness::drive_with_acceptor(
            adapter,
            Arc::clone(&policy),
            Limits::default(),
            Some(Arc::new(acceptor)),
        )
        .await
        .expect("the loopback mounts");

        let result =
            authenticate_publickey(&mut session.client, "tunnel-rider", fixture("alpha")).await;
        assert!(
            result.success(),
            "the fixture identity authenticates before every -R row"
        );
        (session, receiver)
    }

    /// A free loopback port (the rows' pick helper; the listen binds
    /// exactly what it was asked).
    async fn pick_port() -> u16 {
        let probe = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the probe binds");
        probe.local_addr().expect("addr").port()
    }

    /// The arrival row's pump legs: the arrival's request reaches the
    /// forwarded SSH channel, the channel's reply reaches the arrival
    /// socket: the fabric `BiStream` ↔ SSH channel duplex, zero
    /// framing both directions (the route-1 bridge's primitive,
    /// reversed).
    async fn arrival_pump(
        arrival: &mut tokio::net::TcpStream,
        channel: &mut Channel<Msg>,
        request: &[u8],
        reply: &[u8],
        context: &str,
    ) {
        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < request.len() {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("the forwarded channel ended before the request arrived"),
                }
            }
        })
        .await
        .expect("the arrival's request reaches the forwarded channel in time");
        assert_eq!(
            seen, request,
            "{context}: the fabric → SSH pump carries the arrival bytes unframed"
        );

        channel
            .data_bytes(reply.to_vec())
            .await
            .expect("{context}: the SSH channel carries the reply toward the arrival");
        let mut seen_reply = vec![0u8; reply.len()];
        tokio::time::timeout(Duration::from_secs(15), arrival.read_exact(&mut seen_reply))
            .await
            .expect("the reply reaches the arrival socket in time")
            .expect("{context}: the SSH → fabric pump carries the reply bytes unframed");
        assert_eq!(seen_reply, reply, "{context}: the reply arrives verbatim");
    }

    /// - *Contract stamp:* mapping.md §Global-request rows
    ///   (`tcpip_forward` → the family-ward listen dial) + §Channel-type
    ///   rows, the `forwarded-tcpip` row (the data leg: per fabric
    ///   arrival the door opens a `forwarded-tcpip` channel **toward
    ///   the vanilla client**, echoing the listen request's own args
    ///   (the OpenSSH matcher parity) and pumping the fabric
    ///   `BiStream` ↔ the SSH channel). Observed end to end through the
    ///   harness's accept hook.
    #[tokio::test]
    async fn tcpip_forward_serves_through_the_family_listen() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let port = pick_port().await;
        // A nonzero request's success reply carries no port (russh/sshd
        // parity); the arrival's echo below is the actually-bound
        // assertion.
        let _reply = tcpip_forward(&session.client, "127.0.0.1", u32::from(port))
            .await
            .expect("the featured + configured listen serves");

        let mut arrival = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("the fabric's listener accepts the arrival");
        arrival
            .write_all(b"arrival-request")
            .await
            .expect("the arrival socket carries the request");

        let open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the forwarded open arrives in time")
            .expect("the acceptor observed the server-initiated open");
        let ForwardedOpen::Tcpip {
            connected_address,
            connected_port,
            originator_address,
            originator_port,
            mut channel,
        } = open
        else {
            panic!("the -R data leg opens forwarded-tcpip, got another open shape");
        };
        assert_eq!(
            connected_address, "127.0.0.1",
            "the connected address echoes the listen request's wire string verbatim (sshd parity)"
        );
        assert_eq!(
            connected_port,
            u32::from(port),
            "the port echoes the actually-bound number (the port the fabric bound)"
        );
        assert_eq!(
            originator_address, "127.0.0.1",
            "the originator carries the arrival's peer (TunnelForwardedParams.peer)"
        );
        assert!(
            originator_port > 0,
            "the originator's port is the arrival's ephemeral port"
        );

        arrival_pump(
            &mut arrival,
            &mut channel,
            b"arrival-request",
            b"forwarded-reply",
            "the tcpip_forward serve row",
        )
        .await;

        session.unmount().await;
    }

    /// - *Contract stamp:* `client.md` §Lifecycle primitives (the
    ///   bound-port spec: `port = 0` → the server picks and the
    ///   actually-bound number returns); the out-param carries the
    ///   fabric's `bound` reply field's port, and the arrival's echo
    ///   port is that same allocated number (sshd's `get_local_port`
    ///   parity: a forwarded open for a `0`-minted listener never
    ///   claims port 0).
    #[tokio::test]
    async fn port_zero_mint_is_resolved_into_the_bound_port_out_param() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let bound = tcpip_forward(&session.client, "127.0.0.1", 0)
            .await
            .expect("the port-0 mint serves");
        assert_ne!(bound, 0, "the fabric's actually-bound port rides out");

        let mut arrival =
            tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("u16")))
                .await
                .expect("the minted listener accepts");
        arrival.write_all(b"zero-mint").await.expect("rides");

        let open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the forwarded open arrives in time")
            .expect("the acceptor observed the open");
        let ForwardedOpen::Tcpip {
            connected_address,
            connected_port,
            mut channel,
            ..
        } = open
        else {
            panic!("the tcp forward opened forwarded-tcpip");
        };
        assert_eq!(connected_address, "127.0.0.1");
        assert_eq!(connected_port, bound, "the echo port is the minted one");
        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < 9 {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("the forwarded channel ended before the arrival rode"),
                }
            }
        })
        .await
        .expect("the pump carries the arrival in time");
        assert_eq!(seen, b"zero-mint");

        session.unmount().await;
    }

    /// - *Contract stamp:* the cancel forms (`cancel_tcpip_forward`):
    ///   1:1 into the family listener's teardown; the matching listen
    ///   cancels `Ok`, the fabric's bound listener goes away (a
    ///   re-connect is refused), and a second cancel of the same
    ///   (address, port) pair answers the not-found denial
    ///   (`RequestDenied`, the sshd cancel-of-unknown parity).
    #[tokio::test]
    async fn cancel_tcpip_forward_tears_the_fabric_listener_down() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let port = pick_port().await;
        let bound = u32::from(port);
        tcpip_forward(&session.client, "127.0.0.1", bound)
            .await
            .expect("the listen serves before its cancel");

        cancel_tcpip_forward(&session.client, "127.0.0.1", bound)
            .await
            .expect("the pending listen's cancel answers success");
        let denied = cancel_tcpip_forward(&session.client, "127.0.0.1", bound)
            .await
            .expect_err("a second cancel of the same pair is the not-found denial");
        assert!(
            matches!(denied, Error::RequestDenied),
            "expected RequestDenied, got {denied:?}"
        );

        // The fabric listener is gone (the teardown cascaded: the
        // door's listener session dropped, the far accept loop's EOF
        // sentinel fired, the plan's listener dropped, the bind
        // releases). Bounded retry: the connect is refused.
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                match tokio::net::TcpStream::connect(("127.0.0.1", port)).await {
                    Err(_) => return,
                    Ok(stream) => drop(stream),
                }
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        })
        .await
        .expect("the fabric listener never came back after the cancel");

        // The torn-down listen could not have served a late arrival as
        // an SSH channel: nothing rode the acceptor.
        tokio::time::timeout(Duration::from_millis(500), opens.recv())
            .await
            .expect_err("no forwarded open may follow a cancelled listen");
        assert!(opens.is_empty());

        session.unmount().await;
    }

    /// - *Contract stamp:* the wave-3 gate's booked finding item 4
    ///   (`w4-router-collision`): the `ListenRouter` token-collision
    ///   pre-check. The fabric's ADR-008 mint discipline is the
    ///   upstream guarantee the door defends in depth against: the
    ///   test forces what the discipline prevents (the double
    ///   re-mints a live token), and the second listen **refuses
    ///   fail-closed** (`RequestDenied`), its minted fabric listener
    ///   torn down immediately (the refused registration never left a
    ///   half state), the listener cap released (the next mint is
    ///   admitted against the same 1-count ceiling), and the **first
    ///   route still serves**: the collided mint can never steal a
    ///   live adapter's arrival routes.
    #[tokio::test]
    async fn a_collided_mint_is_refused_and_never_steals_the_live_route() {
        let double = wire_fabric_with_listen_mint(
            true,
            true,
            MintStrategy::WithCollisions {
                token: 7,
                collisions: 2,
            },
        )
        .await;
        // The ceiling of 2 both admits the collided mint past the
        // listener cap (so the refusal is the route register, not the
        // cap, the cap row's shape) and makes the refusal's cap
        // release observable: without the release, the third mint
        // would be cap-exceeded.
        let limits = Limits {
            max_concurrent_channels: 2,
            ..Limits::default()
        };
        let (session, mut opens) = listen_session(&double, limits).await;

        let port_a = pick_port().await;
        tcpip_forward(&session.client, "127.0.0.1", u32::from(port_a))
            .await
            .expect("the first listen registers its route normally");

        // The second mint deliberately re-mints the live token: the
        // route registration refuses, the listen declines fail-closed.
        let port_b = pick_port().await;
        let refused = tcpip_forward(&session.client, "127.0.0.1", u32::from(port_b))
            .await
            .expect_err("the collided mint's listen refuses");
        assert!(
            matches!(refused, Error::RequestDenied),
            "expected RequestDenied, got {refused:?}"
        );

        // The collided listen's minted listener was torn down
        // immediately: its bound endpoint never serves (bounded retry,
        // the cancel row's teardown-cascade shape).
        tokio::time::timeout(Duration::from_secs(10), async {
            loop {
                match tokio::net::TcpStream::connect(("127.0.0.1", port_b)).await {
                    Err(_) => return,
                    Ok(stream) => drop(stream),
                }
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
        })
        .await
        .expect("the refused listen's fabric listener was torn down");

        // The listener count was released at the refusal: against the
        // same 2-count ceiling, the next mint (a fresh token, since
        // collision was two mints on one listen, not a broken fabric)
        // is admitted and registers normally (a `0` mint's success
        // reply carries the actually-bound number, the sshd parity).
        let third = tcpip_forward(&session.client, "127.0.0.1", 0)
            .await
            .expect("the collision released the listener count");
        assert_ne!(third, 0);

        // The first route still serves; the steal is impossible: an
        // arrival onto the live route's endpoint opens the
        // forwarded-tcpip channel and the bytes ride the pump.
        let mut arrival = tokio::net::TcpStream::connect(("127.0.0.1", port_a))
            .await
            .expect("the live route's listener still accepts arrivals");
        let request = b"still-served";
        arrival
            .write_all(request)
            .await
            .expect("the arrival's bytes ride toward the door");
        let open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the live route's arrival open arrives in time")
            .expect("observed");
        let ForwardedOpen::Tcpip { mut channel, .. } = open else {
            panic!("the live route's arrival opened something else");
        };
        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < request.len() {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("the live route's channel ended before the request rode"),
                }
            }
        })
        .await
        .expect("the live route's pump rides in time");
        assert_eq!(seen, request, "the collided mint stole nothing");

        // Nothing rode the acceptor beyond the live route's served
        // arrival (the refused mint's listener was torn down at birth
        // so its arrivals could not exist).
        tokio::time::timeout(Duration::from_millis(500), opens.recv())
            .await
            .expect_err("no arrival may follow the refused mint");
        assert!(opens.is_empty());

        session.unmount().await;
    }

    /// - *Contract stamp:* the forwarded gate's miss class (alktunnels
    ///   0.5.0, review 003 W2 adoption) at the alkssh level: an arrival
    ///   whose `listen` token has no route (nothing was ever minted for
    ///   this door) fails the open at the call plane as a typed
    ///   `channel:open_failed` (`TunnelEstablishError::UnknownResource`
    ///   the wire's `unknown_resource` reason) **before any channel
    ///   is born**; the phantom-channel shape (call-plane success,
    ///   then handler-side decline) is closed at the open: no channel
    ///   allocation, no adopted halves, no acceptor row.
    #[tokio::test]
    async fn a_correlation_miss_fails_the_open_typed_before_any_channel() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let params = alktunnels::params::TunnelForwardedParams {
            listen: 424_242,
            substrate: alktunnels::params::Substrate::Tcp,
            peer: alktunnels::params::SubstrateAddr::HostPort {
                host: "127.0.0.1".to_owned(),
                port: 1234,
            },
        };
        let outcome = double.open_forwarded_directly(params).await;
        let err = match outcome {
            Ok(_) => {
                panic!("the gate miss fails the open typed; no phantom channel exists to adopt")
            }
            Err(error) => error,
        };
        assert!(
            matches!(err, TunnelEstablishError::UnknownResource(_)),
            "expected UnknownResource at the call plane, got {err:?}"
        );

        // No arrival rode the acceptor (the open died at the call
        // plane; the handler never ran.
        tokio::time::timeout(Duration::from_millis(500), opens.recv())
            .await
            .expect_err("a gate-missed arrival never opens an SSH channel");
        assert!(opens.is_empty());

        session.unmount().await;
    }

    /// - *Contract stamp:* the forwarded gate's pass posture (alktunnels
    ///   0.5.0, review 003 W2 adoption): an arrival whose token is
    ///   genuinely routed behaves exactly as the base shape did; the
    ///   open succeeds at the call plane, the door allocates the SSH
    ///   channel at the acceptor, and the fabric-side halves are the
    ///   arrival's data plane (the gate adds an establishment phase;
    ///   the serving flow is unchanged).
    #[tokio::test]
    async fn a_gate_pass_arrival_serves_the_route_unchanged() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let port = pick_port().await;
        let _bound = tcpip_forward(&session.client, "127.0.0.1", u32::from(port))
            .await
            .expect("the listen is minted");
        let token = double.minted_token().await;

        let params = alktunnels::params::TunnelForwardedParams {
            listen: token,
            substrate: alktunnels::params::Substrate::Tcp,
            peer: alktunnels::params::SubstrateAddr::HostPort {
                host: "127.0.0.1".to_owned(),
                port: 1234,
            },
        };
        let (mut send, _recv) = double
            .open_forwarded_directly(params)
            .await
            .expect("the gate-passed arrival adopts its fabric-side halves");

        // The serving flow is the unchanged shape: the arrival becomes
        // an SSH `forwarded-tcpip` open at the acceptor, and data rides
        // the fabric half through to it.
        let open = tokio::time::timeout(Duration::from_secs(10), opens.recv())
            .await
            .expect("the gate-passed arrival observed in time")
            .expect("the acceptor's stream stays alive");
        let mut channel = match open {
            ForwardedOpen::Tcpip { channel, .. } => channel,
            other => panic!("the gate-passed arrival opened something else: {other:?}"),
        };

        send.write_all(b"orphan arrival")
            .await
            .expect("the adopted halves still write (the serving flow ran)");
        drop(send);

        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < 13 {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => return,
                }
            }
        })
        .await
        .expect("the fabric half rides in time");
        assert_eq!(
            seen,
            b"orphan arrival".to_vec(),
            "the gate-passed arrival's data rode the unchanged serving flow"
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* AGENTS.md convention 7 (the visible-surface
    ///   invariant's alkcall half on the `-R` leg): the door-side
    ///   registry's `AccessControl::check` gates arrivals by the fabric
    ///   caller's identity (an identity lacking the
    ///   `tunnel:forwarded` scope is refused by alkcall's dispatch (the
    ///   openable's own spec) and no SSH channel opens, even for a
    ///   genuinely minted token. The genuinely-minted-token shape
    ///   double-pins the gate ordering now (the establisher variant's
    ///   pinned order, alktunnels 0.5.0 / ADR-008 Amendment 2): the ACL
    ///   refuses before the establisher's gate ever looks up the token;
    ///   an unscoped arrival could not leak a gate pass.
    #[tokio::test]
    async fn the_door_side_acl_refuses_an_unscoped_arrival() {
        let double = wire_fabric_with_listen(true, false).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let port = pick_port().await;
        let _bound = tcpip_forward(&session.client, "127.0.0.1", u32::from(port))
            .await
            .expect(
                "the listen itself is gated by the fabric-side identity, which carries the scope",
            );

        // A real token (minted for this listener), unscoped arrival
        // identity: the door-side ACL refuses the open op, and the fabric
        // side sees the refusal, never a channel allocation.
        let token = double.minted_token().await;
        let params = alktunnels::params::TunnelForwardedParams {
            listen: token,
            substrate: alktunnels::params::Substrate::Tcp,
            peer: alktunnels::params::SubstrateAddr::HostPort {
                host: "127.0.0.1".to_owned(),
                port: 4321,
            },
        };
        let outcome = double.open_forwarded_directly(params).await;
        assert!(
            outcome.is_err(),
            "the unscoped arrival identity is refused at the door's dispatch"
        );

        // Nothing rode the acceptor.
        tokio::time::timeout(Duration::from_millis(500), opens.recv())
            .await
            .expect_err("an ACL-refused arrival never opens an SSH channel");
        assert!(opens.is_empty());

        session.unmount().await;
    }

    /// - *Contract stamp:* AGENTS.md convention 9 (the listener count
    ///   is a bounded resource): the per-session listener map is
    ///   bounded through its dedicated counter against the same
    ///   `max_concurrent_channels` ceiling; the listen beyond the
    ///   count answers the vanilla `RequestDenied` while channels stay
    ///   free, and the cancel releases the count again.
    #[tokio::test]
    async fn the_listener_cap_bounds_the_session_listen_map() {
        let double = wire_fabric_with_listen(true, true).await;
        let limits = Limits {
            max_concurrent_channels: 1,
            ..Limits::default()
        };
        let (session, mut opens) = listen_session(&double, limits).await;

        let first = tcpip_forward(&session.client, "127.0.0.1", 0)
            .await
            .expect("the first listen is admitted");
        assert_ne!(first, 0);
        let err = tcpip_forward(&session.client, "127.0.0.1", 0)
            .await
            .expect_err("the second listen rides beyond the listener cap");
        assert!(
            matches!(err, Error::RequestDenied),
            "expected RequestDenied, got {err:?}"
        );

        // The cap recovery: the cancel of the first listen frees the
        // count; the next mint is admitted.
        cancel_tcpip_forward(&session.client, "127.0.0.1", first)
            .await
            .expect("the cancel releases the count");
        let second = tcpip_forward(&session.client, "127.0.0.1", 0)
            .await
            .expect("the listener count recovered at the cancel");
        assert_ne!(second, 0);

        // Nothing in this row ever manufactured an arrival.
        tokio::time::timeout(Duration::from_millis(300), opens.recv())
            .await
            .expect_err("no arrivals were manufactured in this row");
        assert!(opens.is_empty());

        session.unmount().await;
    }

    /// - *Contract stamp:* AGENTS.md convention 9 / the acceptor
    ///   vocabulary's cap rows (the wave-2 gate's booked shape): the
    ///   per-arrival admission is the owning session's in-crate
    ///   channel cap: an arrival onto an exhausted count is refused
    ///   at the accept surface (no SSH channel opens; the fabric half
    ///   drops, teardown-at-birth), and an established channel's close
    ///   recovers the count (the next arrival is admitted, the
    ///   release proof).
    #[tokio::test]
    async fn the_arrival_admission_is_cap_gated_at_the_accept_surface() {
        let double = wire_fabric_with_listen(true, true).await;
        let limits = Limits {
            max_concurrent_channels: 1,
            ..Limits::default()
        };
        let (session, mut opens) = listen_session(&double, limits).await;

        let port = pick_port().await;
        let _reply = tcpip_forward(&session.client, "127.0.0.1", u32::from(port))
            .await
            .expect("a minted listen costs no channel count");

        // The first arrival admits itself: its SSH channel opens and
        // the count is full (held by the arrival socket + channel,
        // both deliberately kept alive un-served).
        let first_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("the first arrival is admitted");
        let first_open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the first arrival's open arrives")
            .expect("observed");
        let ForwardedOpen::Tcpip {
            channel: first_channel,
            ..
        } = first_open
        else {
            panic!("the tcp arrival opened forwarded-tcpip");
        };

        // The second arrival meets the full count: no SSH channel
        // opens (its fabric half teardown-at-births), and its socket
        // EOFs bounded.
        let mut second_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("the fabric listener still accepts arrivals");
        tokio::time::timeout(Duration::from_millis(700), opens.recv())
            .await
            .expect_err("an arrival beyond the cap never opens an SSH channel");
        let eof = tokio::time::timeout(Duration::from_secs(10), second_arrival.read_u8())
            .await
            .expect("the refused arrival's teardown is observable in time")
            .expect_err("the refused arrival's socket EOFs");
        let _ = eof;

        // The release proof: the first arrival's socket goes; the
        // pump's fabric→SSH direction EOFs (the pump's
        // shutdown-on-completion half-closes the SSH channel; the
        // OpenSSH client, whose target socket died with the arrival,
        // closes its half; the harness's stand-in does exactly that).
        // That close (not the pump) is what fires the packet loop's
        // established-channel release.
        drop(first_arrival);
        let _ = first_channel.close().await;

        let mut third_arrival = tokio::net::TcpStream::connect(("127.0.0.1", port))
            .await
            .expect("the fabric listener still serves");
        third_arrival.write_all(b"again").await.expect("rides");
        let third_open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the next arrival's open arrives; the released count recovered")
            .expect("observed");
        let ForwardedOpen::Tcpip {
            channel: mut third_channel,
            ..
        } = third_open
        else {
            panic!("the tcp arrival opened forwarded-tcpip");
        };
        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < 5 {
                match third_channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("the forwarded channel ended before the bytes rode"),
                }
            }
        })
        .await
        .expect("the pump rides in time");
        assert_eq!(seen, b"again");

        session.unmount().await;
    }

    /// - *Contract stamp:* OQ-11 on the listen flow (the runtime-Unix
    ///   gate riding the `streamlocal_forward` serving flow) + the unix
    ///   row of the `-R` leg: the unix listen serves through the fabric
    ///   (the path echoes verbatim; no port semantics, no port
    ///   out-param), the arrival opens
    ///   `forwarded-streamlocal@openssh.com` toward the client, and the
    ///   cancel form tears it down 1:1 (path-matched).
    #[cfg(unix)]
    #[tokio::test]
    async fn streamlocal_forward_serves_arrivals_and_cancels_through_the_family() {
        let double = wire_fabric_with_listen(true, true).await;
        let (session, mut opens) = listen_session(&double, Limits::default()).await;

        let path = format!("/tmp/alkssh-w3-listen-{}-listen.sock", std::process::id());
        super::common::harness::streamlocal_forward(&session.client, &path)
            .await
            .expect("the unix listen serves");

        let mut arrival = tokio::net::UnixStream::connect(&path)
            .await
            .expect("the fabric's unix listener accepts the arrival");
        arrival.write_all(b"unix-arrival").await.expect("rides");

        let open = tokio::time::timeout(Duration::from_secs(15), opens.recv())
            .await
            .expect("the forwarded open arrives in time")
            .expect("observed");
        let ForwardedOpen::Streamlocal {
            socket_path,
            mut channel,
        } = open
        else {
            panic!("the unix arrival opens forwarded-streamlocal");
        };
        assert_eq!(
            socket_path, path,
            "the server_socket_path echoes the listen request's path verbatim"
        );
        let mut seen: Vec<u8> = Vec::new();
        tokio::time::timeout(Duration::from_secs(15), async {
            while seen.len() < 12 {
                match channel.wait().await {
                    Some(ChannelMsg::Data { data }) => seen.extend_from_slice(&data),
                    Some(_) => continue,
                    None => panic!("the forwarded channel ended before the arrival rode"),
                }
            }
        })
        .await
        .expect("the unix pump carries the arrival");
        assert_eq!(seen, b"unix-arrival");
        channel
            .data_bytes(b"unix-reply".to_vec())
            .await
            .expect("the SSH channel carries the reply");
        let mut reply = vec![0u8; 10];
        tokio::time::timeout(Duration::from_secs(15), arrival.read_exact(&mut reply))
            .await
            .expect("the reply reaches the arrival stream in time")
            .expect("rides");
        assert_eq!(reply, b"unix-reply");

        super::common::harness::cancel_streamlocal_forward(&session.client, &path)
            .await
            .expect("the pending unix listen's cancel answers success");
        let denied = super::common::harness::cancel_streamlocal_forward(&session.client, &path)
            .await
            .expect_err("a second cancel of the same path is the not-found denial");
        assert!(matches!(denied, Error::RequestDenied), "got {denied:?}");

        session.unmount().await;
        let _ = std::fs::remove_file(&path);
    }

    /// - *Contract stamp:* the unset-declines contract on the listen
    ///   seam (`w3-tunnel-listen`'s whole-`Option` shape; an
    ///   egress-only door is legitimate config): a `tunnel` build with
    ///   a fabric leg but no `tunnel_listen_seam` declines every listen
    ///   form with the vanilla `RequestDenied`, while the leg-1 rows
    ///   still route (the egress bridge is not dead).
    #[tokio::test]
    async fn listen_forms_decline_when_the_listen_seam_is_unset() {
        let targets: TargetMap = Arc::default();
        let client = wire_fabric(
            inproc_dial(targets, Arc::new(AtomicUsize::new(0))),
            scoped_caller(),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default());
        serve.channel_client = Some(client);
        let adapter = SshAdapter::new(serve)
            .await
            .expect("the egress-only door constructs without the seam");

        let policy = Arc::new(MemHostKeyPolicy::with_default(true));
        let mut session =
            super::common::harness::drive(adapter, Arc::clone(&policy), Limits::default())
                .await
                .expect("the loopback mounts");
        let result =
            authenticate_publickey(&mut session.client, "tunnel-rider", fixture("alpha")).await;
        assert!(result.success());

        let err = tcpip_forward(&session.client, "127.0.0.1", 8080)
            .await
            .expect_err("the unset seam declines the listen");
        assert!(matches!(err, Error::RequestDenied), "got {err:?}");
        let err = cancel_tcpip_forward(&session.client, "127.0.0.1", 8080)
            .await
            .expect_err("the unset seam declines the cancel too");
        assert!(matches!(err, Error::RequestDenied), "got {err:?}");

        // The egress door still routes: the direct row reaches the
        // fabric (the fail-dial class `ConnectFailed`, not
        // `AdministrativelyProhibited`).
        let err = open_direct_tcpip(&session.client, "127.0.0.1", 36005)
            .await
            .expect_err("the direct row still routes onto the bridge");
        open_failure(
            err,
            ChannelOpenFailure::ConnectFailed,
            "the egress-only door's direct row (the seam-unconfigured row's control leg)",
        );

        session.unmount().await;
    }

    /// - *Contract stamp:* the door-side registration is
    ///   one-per-op-per-registry (mapping.md §Global-request rows / the
    ///   `install_forwarded_surface` claim): alkcall's `register`
    ///   silently replaces an existing op entry, so the door
    ///   pre-checks; a second `SshAdapter` composed over the same
    ///   shared seam registry is a **failed construction**
    ///   (`AdapterError::ListenRegistration`), never a silent steal of
    ///   the first adapter's arrival routes.
    #[tokio::test]
    async fn a_second_adapter_over_a_shared_listen_registry_fails_construction() {
        let double = wire_fabric_with_listen(true, true).await;
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );

        let mut first = harness_serve_config(
            Arc::clone(&provider) as Arc<dyn alkcall::core::auth::IdentityProvider>,
            &[fixture("bravo")],
            Limits::default(),
        );
        first.channel_client = Some(Arc::clone(&double.client));
        first.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let _first = SshAdapter::new(first)
            .await
            .expect("the first adapter over the shared registry constructs");

        let mut second = harness_serve_config(provider, &[fixture("charlie")], Limits::default());
        second.channel_client = Some(Arc::clone(&double.client));
        second.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let refused = SshAdapter::new(second)
            .await
            .expect_err("a conflicting registration is a failed construction");
        assert!(
            matches!(refused, alkssh::AdapterError::ListenRegistration(_)),
            "the refusal is the ListenRegistration class, got {refused:?}"
        );
    }
}

#[cfg(feature = "tunnel")]
mod client_smoke {
    //! The full-circle smoke (`w4-reverse-listen`'s recorded posture):
    //! the **alkssh client's** `-R` rows against the **alkssh server
    //! direction's** `-R` rows, both directions of the crate in one
    //! live wire, the route-1 bridge's two legs meeting through the
    //! alkcall fabric double (the composition the alkforge assembly
    //! makes). A connection to the minted bound port round-trips
    //! through: the fabric's real TCP listener → the fabric's
    //! bind-first accept loop → the door-side `tunnel/forwarded`
    //! arrival → the door's server-initiated `forwarded-tcpip` open
    //! (echo args + originator) → the alkssh client's acceptor registry
    //! → the acceptor duplex → the echo back. Real-`sshd` `-R` is
    //! wave-5 e2e territory (never silently dropped).

    use std::sync::Arc;
    use std::time::Duration;

    use async_trait::async_trait;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    use alkssh::{
        AcceptorRegistry, DialConfig, ForwardedTcpip, TcpipForwardedAcceptor, TunnelListenSeam,
        dial,
    };

    use super::common::harness::serve_config as harness_serve_config;
    use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
    use super::serving::wire_fabric_with_listen;

    struct EchoAcceptor {
        opens: tokio::sync::mpsc::UnboundedSender<(String, u32, String)>,
    }

    #[async_trait]
    impl TcpipForwardedAcceptor for EchoAcceptor {
        async fn accept(&self, open: ForwardedTcpip) {
            let _ = self.opens.send((
                open.connected_address.clone(),
                open.connected_port,
                open.originator_address.clone(),
            ));
            tokio::spawn(async move {
                let mut stream = open.stream;
                let mut buf = vec![0u8; 4096];
                loop {
                    match stream.read(&mut buf).await {
                        Ok(0) | Err(_) => return,
                        Ok(count) => {
                            if stream.write_all(&buf[..count]).await.is_err() {
                                return;
                            }
                        }
                    }
                }
            });
        }
    }

    #[tokio::test]
    async fn smoke_client_reverse_listen_full_circle_through_the_server_direction() {
        let double = wire_fabric_with_listen(true, true).await;

        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("reverse-runner", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Default::default());
        serve.channel_client = Some(Arc::clone(&double.client));
        serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let adapter = alkssh::SshAdapter::new(serve)
            .await
            .expect("the -R wired adapter constructs");

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the smoke's loopback listener binds");
        let addr = listener.local_addr().expect("the smoke has an address");
        tokio::spawn(async move {
            loop {
                let Ok((socket, _)) = listener.accept().await else {
                    break;
                };
                if adapter.handle(socket).await.is_err() {
                    break;
                }
            }
        });

        let policy = Arc::new(MemHostKeyPolicy::new());
        policy.pin("127.0.0.1", fixture("bravo").fingerprint);
        let config = DialConfig::new(
            addr.to_string(),
            "runner",
            crate_test_key_source(),
            Arc::clone(&policy) as Arc<dyn alkssh::HostKeyPolicy>,
        );
        let connection = dial(config).await.expect("the dial succeeds");

        let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
        let registry: &AcceptorRegistry = connection.acceptors();
        registry
            .install_tcpip(Arc::new(EchoAcceptor { opens: opens_tx }))
            .expect("the acceptor installs");

        let minted_listener = connection
            .reverse_listen("127.0.0.1", 0)
            .await
            .expect("the full-circle mint serves");
        let bound = minted_listener
            .endpoint()
            .bound_port()
            .expect("the tcp mint's handle carries the actually-bound port");
        assert!(
            bound > 0,
            "the server direction's fabric minted a real bound port"
        );

        let mut conn =
            tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("in-range")))
                .await
                .expect("the fabric's listener accepts the arrival");
        conn.write_all(b"full-circle-data")
            .await
            .expect("the arrival carries the bytes");
        let mut echoed = [0u8; 16];
        tokio::time::timeout(Duration::from_secs(30), conn.read_exact(&mut echoed))
            .await
            .expect("the full-circle round trip completes in time")
            .expect("the echo reads");
        assert_eq!(
            &echoed, b"full-circle-data",
            "the bytes returned through every leg: fabric → door arrival \
             → SSH channel → client acceptor duplex → echo → back"
        );

        let (connected_address, connected_port, originator_address) =
            tokio::time::timeout(Duration::from_secs(15), opens_rx.recv())
                .await
                .expect("the acceptor's handoff record arrives")
                .expect("the acceptor sender lives");
        assert_eq!(connected_address, "127.0.0.1");
        assert_eq!(
            connected_port, bound,
            "the echo port is the minted bound one"
        );
        assert_eq!(
            originator_address, "127.0.0.1",
            "the originator carries the fabric arrival's peer"
        );

        let _ = connection.close().await;
        let _ = double.minted_token().await; // the establisher's mint observed (the fabric's bookkeeping)
    }

    /// The smoke's dial key pose (the TestVault posture, client-side).
    fn crate_test_key_source() -> Arc<dyn alkssh::KeySource> {
        let vault = super::common::TestVault::unlocked();
        let blob = vault.seal(fixture("alpha"));
        Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob)))
    }
}

#[cfg(all(feature = "tunnel", unix))]
mod real_tool {
    //! The real-tool smoke (`w3-tunnel-direct`): one `ssh -L` (+ curl)
    //! through a TCP-mounted adapter against the fabric double, the
    //! POC-3 §C.1 gate re-landed. Tooling posture (recorded, like the
    //! git task's): the smoke requires `ssh`/`curl` in the environment;
    //! when absent it is **skipped with a printed reason**
    //! (skipped-with-reason at most, never silently green).

    use std::sync::Arc;

    use alktunnels::params::{Substrate, SubstrateAddr, TUNNEL_DIRECT_SCOPE};
    use alktunnels::producer::{BindFn, BoundListener, DialFn, TargetHandle, TunnelEstablishError};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    use super::common::harness::serve_config as harness_serve_config;
    use super::common::{MemIdentityProvider, fixture, identity};
    use super::serving::{wire_fabric, wire_fabric_with_listen};
    use alkssh::TunnelListenSeam;

    /// The smoke's dial seam: a real-TCP dial closure (hand-rolled;
    /// the assembly's `DialFn` injected over the substrate for real;
    /// alktunnels' `local` feature stays off).
    fn tcp_dial() -> DialFn {
        Arc::new(move |_substrate: Substrate, target: &str| {
            let target = target.to_owned();
            Box::pin(async move {
                let stream = tokio::net::TcpStream::connect(&target).await.map_err(|e| {
                    TunnelEstablishError::DialFailed(format!("dial `{target}`: {e}"))
                })?;
                let (read, write) = stream.into_split();
                Ok(TargetHandle {
                    read: Box::new(read),
                    write: Box::new(write),
                })
            })
        })
    }

    /// A tiny HTTP/1.1 endpoint (the POC's shape), the substrate the
    /// forwarded traffic dials through the fabric.
    async fn spawn_http_endpoint() -> (std::net::SocketAddr, String) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the smoke listener binds");
        let addr = listener.local_addr().expect("local addr");
        let body = "alkssh w3-tunnel-direct through the fabric\n".to_owned();
        let body2 = body.clone();
        tokio::spawn(async move {
            loop {
                let Ok((mut conn, _)) = listener.accept().await else {
                    return;
                };
                let body = body2.clone();
                tokio::spawn(async move {
                    let mut buf = vec![0u8; 8192];
                    let _ = conn.read(&mut buf).await;
                    let head = format!(
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        body.len(),
                        body
                    );
                    let _ = conn.write_all(head.as_bytes()).await;
                    let _ = conn.shutdown().await;
                });
            }
        });
        (addr, body)
    }

    #[tokio::test]
    async fn real_ssh_dash_l_smoke() {
        if std::process::Command::new("ssh")
            .arg("-V")
            .output()
            .is_err()
        {
            eprintln!("SKIPPED (with reason): real-tool tunnel smoke, ssh is unavailable");
            return;
        }
        if std::process::Command::new("curl")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("SKIPPED (with reason): real-tool tunnel smoke, curl is unavailable");
            return;
        }

        let (http_addr, body) = spawn_http_endpoint().await;
        let client = wire_fabric(
            tcp_dial(),
            identity("tunnel-caller", &[TUNNEL_DIRECT_SCOPE]),
            Arc::new(tokio::sync::Mutex::new(None)),
        )
        .await;

        let work = std::env::temp_dir().join(format!("alkssh-tunnel-smoke-{}", std::process::id()));
        std::fs::create_dir_all(&work).expect("the smoke workdir creates");
        let key_path = work.join("id_ed25519");
        std::fs::write(&key_path, fixture("alpha").pem).expect("the fixture key writes");
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600))
            .expect("the key file perms set");

        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );
        let mut serve =
            harness_serve_config(provider, &[fixture("bravo")], alkssh::Limits::default());
        serve.channel_client = Some(client);
        let adapter = Arc::new(
            alkssh::SshAdapter::new(serve)
                .await
                .expect("the smoke adapter constructs"),
        );

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the smoke listener binds");
        let ssh_port = listener.local_addr().expect("local addr").port();
        let accept_task = tokio::spawn(async move {
            loop {
                let Ok((stream, _)) = listener.accept().await else {
                    return;
                };
                let adapter = Arc::clone(&adapter);
                tokio::spawn(async move {
                    let _ = adapter.handle(stream).await;
                });
            }
        });

        let fwd_port: u16 = pick_local_port().await;
        let mut ssh_child = tokio::process::Command::new("ssh")
            .args([
                "-N",
                "-i",
                key_path.to_str().expect("tmp path"),
                "-p",
                &ssh_port.to_string(),
                "-o",
                "BatchMode=yes",
                "-o",
                "IdentitiesOnly=yes",
                "-o",
                "StrictHostKeyChecking=no",
                "-o",
                "UserKnownHostsFile=/dev/null",
                "-o",
                "ConnectTimeout=10",
                "-o",
                "ExitOnForwardFailure=yes",
                "-L",
                &format!(
                    "127.0.0.1:{fwd_port}:{}:{}",
                    http_addr.ip(),
                    http_addr.port()
                ),
            ])
            .arg("tunnel-rider@127.0.0.1")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("vanilla ssh spawns");

        // The -L listener binds client-side; give it a bounded settle
        // window then curl through it.
        let through = tokio::time::timeout(std::time::Duration::from_secs(30), async {
            let mut attempts = 0u32;
            loop {
                let out = tokio::process::Command::new("curl")
                    .args(["-s", "--max-time", "10"])
                    .arg(format!("http://127.0.0.1:{fwd_port}/"))
                    .output()
                    .await
                    .expect("curl spawns");
                if out.status.success() {
                    return out;
                }
                attempts += 1;
                assert!(
                    attempts < 20,
                    "curl never got through: exit {:?}",
                    out.status.code()
                );
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
            }
        })
        .await
        .expect("the smoke resolves in time");

        let _ = ssh_child.kill().await;
        accept_task.abort();

        let stdout = String::from_utf8_lossy(&through.stdout).to_string();
        assert!(
            through.status.success() && stdout.trim_end() == body.trim_end(),
            "vanilla ssh -L + curl through the door → fabric → endpoint returns the body; got {stdout:?}"
        );

        let _ = std::fs::remove_dir_all(&work);
    }

    /// The smoke's bind seam: the leg-2 dynamic-address producer's
    /// real-TCP bind (the wire-named endpoint bound for real; shared
    /// shape with the harness rows' `real_bind`).
    fn tcp_bind() -> BindFn {
        Arc::new(move |_substrate, endpoint: &str| {
            let endpoint = endpoint.to_owned();
            let endpoint = endpoint.replace("localhost:", "127.0.0.1:");
            Box::pin(async move {
                let listener = tokio::net::TcpListener::bind(&endpoint)
                    .await
                    .map_err(|e| {
                        TunnelEstablishError::ResourceShortage(format!("bind `{endpoint}`: {e}"))
                    })?;
                let addr = listener.local_addr().expect("bound addr");
                Ok(BoundListener {
                    listener: Arc::new(listener),
                    bound: SubstrateAddr::HostPort {
                        host: addr.ip().to_string(),
                        port: addr.port(),
                    },
                })
            })
        })
    }

    #[tokio::test]
    async fn real_ssh_dash_r_smoke() {
        if std::process::Command::new("ssh")
            .arg("-V")
            .output()
            .is_err()
        {
            eprintln!("SKIPPED (with reason): real-tool -R smoke, ssh is unavailable");
            return;
        }
        if std::process::Command::new("curl")
            .arg("--version")
            .output()
            .is_err()
        {
            eprintln!("SKIPPED (with reason): real-tool -R smoke, curl is unavailable");
            return;
        }

        // The vanilla client's own dial target (the `-R` convention:
        // the client dials its own endpoint per arrival, from the
        // machine the SSH client runs on).
        let (http_addr, body) = spawn_http_endpoint().await;
        let double = wire_fabric_with_listen(true, true).await;

        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("tunnel-rider", &["ssh"]),
        );
        let mut serve =
            harness_serve_config(provider, &[fixture("bravo")], alkssh::Limits::default());
        serve.channel_client = Some(Arc::clone(&double.client));
        serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let adapter = Arc::new(
            alkssh::SshAdapter::new(serve)
                .await
                .expect("the smoke adapter constructs"),
        );

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the smoke listener binds");
        let ssh_port = listener.local_addr().expect("local addr").port();
        let accept_task = tokio::spawn(async move {
            loop {
                let Ok((stream, _)) = listener.accept().await else {
                    return;
                };
                let adapter = Arc::clone(&adapter);
                tokio::spawn(async move {
                    let _ = adapter.handle(stream).await;
                });
            }
        });

        let fwd_port: u16 = pick_local_port().await;
        let mut ssh_child = tokio::process::Command::new("ssh")
            .args([
                "-N",
                "-i",
                work_key_path().to_str().expect("tmp path"),
                "-p",
                &ssh_port.to_string(),
                "-o",
                "BatchMode=yes",
                "-o",
                "IdentitiesOnly=yes",
                "-o",
                "StrictHostKeyChecking=no",
                "-o",
                "UserKnownHostsFile=/dev/null",
                "-o",
                "ConnectTimeout=10",
                "-o",
                "ExitOnForwardFailure=yes",
                "-R",
                &format!(
                    "127.0.0.1:{fwd_port}:{}:{}",
                    http_addr.ip(),
                    http_addr.port()
                ),
            ])
            .arg("tunnel-rider@127.0.0.1")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("vanilla ssh spawns");

        // The fabric bound the -R listener; curl arrives on it, the
        // door maps the arrival to a forwarded-tcpip channel the
        // vanilla client accepts and dials its own endpoint through.
        let through = tokio::time::timeout(std::time::Duration::from_secs(30), async {
            let mut attempts = 0u32;
            loop {
                let out = tokio::process::Command::new("curl")
                    .args(["-s", "--max-time", "10"])
                    .arg(format!("http://127.0.0.1:{fwd_port}/"))
                    .output()
                    .await
                    .expect("curl spawns");
                if out.status.success() {
                    return out;
                }
                attempts += 1;
                assert!(
                    attempts < 20,
                    "curl never got through: exit {:?}",
                    out.status.code()
                );
                tokio::time::sleep(std::time::Duration::from_millis(250)).await;
            }
        })
        .await
        .expect("the -R smoke resolves in time");

        let _ = ssh_child.kill().await;
        accept_task.abort();

        let stdout = String::from_utf8_lossy(&through.stdout).to_string();
        assert!(
            through.status.success() && stdout.trim_end() == body.trim_end(),
            "vanilla ssh -R + curl through the fabric → arrival → forwarded-tcpip → client-dialed endpoint returns the body; got {stdout:?}"
        );
    }

    /// The smoke's per-run workdir key path (the -L smoke's inline
    /// shape extracted).
    fn work_key_path() -> std::path::PathBuf {
        use std::os::unix::fs::PermissionsExt;

        let work =
            std::env::temp_dir().join(format!("alkssh-tunnel-r-smoke-{}", std::process::id()));
        std::fs::create_dir_all(&work).expect("the smoke workdir creates");
        let key_path = work.join("id_ed25519");
        std::fs::write(&key_path, fixture("alpha").pem).expect("the fixture key writes");
        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600))
            .expect("the key file perms set");
        key_path
    }

    async fn pick_local_port() -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the probe binds");
        let port = listener.local_addr().expect("addr").port();
        drop(listener);
        port
    }
}