//! The client-direction real-tool e2e (`w5-e2e-client`) — the
//! reverse-runner shape re-landed against a **real `sshd`**
//! (`docs/plans/implementation.md` §Wave 5, the real-tool e2e bullet's
//! client half). The wave-4 client rows ran against the in-process
//! fake `sshd` double (`tests/common/sshd_double.rs`); the wave-4 gate
//! named real-`sshd` e2e wave-5 territory explicitly. This binary
//! dials a spawned OpenSSH server with the shipped crate: exec
//! (exit-status parity through a real server's reaping), the
//! direct-tcpip dial (`-L`-shaped), the `-R` leg (the port-0 mint
//! against a real server's actually-bound reply, the server-initiated
//! `forwarded-tcpip` open served through the acceptor registry — the
//! reverse-runner consumer pattern end to end), the cancel form
//! tearing the real listen down, and the decline rows (the sftp
//! client-side placeholder decline, the server-refused direct-tcpip
//! open with the connection staying usable, the host-key policy
//! mismatch hard-decline — the case-1 pin posture against a real
//! presented key).
//!
//! **Placement decision (recorded, the task's acceptance criterion
//! 1):** a dedicated `tests/e2e_client.rs` binary reusing
//! `tests/common` — the wave-4 client binaries (`tests/client_dial.rs`
//! / `client_ops.rs` / `client_reverse_listen.rs`) keep their
//! single-writer posture for `w5-decline-matrix`; this binary owns
//! only its own rows and its in-file `support` fixture module,
//! disjoint from `tests/e2e_server.rs` (the matrix task's binaries).
//!
//! **The sshd fixture shape (decided + recorded):** a **per-row
//! spawned** real `sshd` — each row's config knobs differ
//! (`AllowTcpForwarding` yes/no) and each row dials its own server,
//! so a shared per-binary instance buys nothing: per-row isolation
//! (own workdir, own minted ed25519 host key, own probed port) beats
//! the spawn cost. The knob set: publickey-only
//! (`PubkeyAuthentication yes`, password + keyboard-interactive off),
//! `StrictModes no` (the fixture's throwaway workdir carries none of
//! the real home-dir hygiene), `UsePAM no`, `UseDNS no`, `PidFile
//! none`, the minted host key + the fixture-`alpha`-authorized
//! `authorized_keys` at absolute paths, and per-row
//! `AllowTcpForwarding`. Readiness is a bounded TCP probe against the
//! dialed port; a server that never comes up (binary absent,
//! `ssh-keygen` absent, spawn refused, privilege-separation directory
//! missing, config refused) is the row's **reasoned skip** with the
//! server's own `-e` log as the reason — never silently green. The
//! dial rides the decrypt-at-dial path over the in-memory vault
//! double (`VaultKeySource` over `TestVault`, fixture `alpha` sealed),
//! and the case-1 pin posture (`MemHostKeyPolicy::pin`) accepts only
//! the fixture's minted host fingerprint.
//!
//! Rows are `#[cfg(unix)]` (the fixture spawns processes) and
//! feature-gated per row group under the established check-gated
//! **skipped-with-reason** posture. Every real-tool interaction is
//! deadline-bounded — a decline must surface a typed value to the
//! dialer, never a hang. Rows carry `Contract stamp:` doc markers
//! against `client.md` / `reverse-runner.md` (the greppable
//! `tests/common` discipline).

mod common;

#[cfg(unix)]
mod support {
    #![allow(dead_code)]

    use std::io::Read;
    use std::path::{Path, PathBuf};
    use std::process::Command;
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::{DialConfig, DialError, HostKeyPolicy, SshConnection, VaultKeySource, dial};

    use super::common::{MemHostKeyPolicy, TestVault, fixture};

    /// One shared row deadline — every real-process interaction is
    /// bounded (the POC-1 §5 posture: a decline never hangs).
    pub const ROW_DEADLINE: Duration = Duration::from_secs(30);

    /// The sshd candidates probed in order (`sbin` paths first — the
    /// server binary rarely sits on the user's `PATH`).
    const SSHD_CANDIDATES: &[&str] = &[
        "/usr/sbin/sshd",
        "/usr/local/sbin/sshd",
        "/sbin/sshd",
        "/usr/bin/sshd",
    ];

    /// Prints the row's reasoned skip (never silently green).
    pub fn skip(row: &str, reason: &str) {
        eprintln!("SKIPPED (with reason): {row} — {reason}");
    }

    /// The `sshd` binary's location — probed once, `None` when the
    /// environment carries no server (every row's reasoned skip).
    pub fn sshd_binary() -> Option<PathBuf> {
        SSHD_CANDIDATES
            .iter()
            .map(PathBuf::from)
            .find(|path| path.is_file())
    }

    /// True when spawning the probe succeeds at all (the minting
    /// tool's own exit codes are its business).
    pub fn keygen_present() -> bool {
        Command::new("ssh-keygen").arg("-h").output().is_ok()
    }

    /// The per-row fixture workdir — a throwaway directory under the
    /// temp dir, one per row (the wave-5 e2e-server lesson: no shared
    /// workdirs, no cross-row races).
    fn workdir(row: &str) -> PathBuf {
        let work =
            std::env::temp_dir().join(format!("alkssh-e2e-client-{row}-{}", std::process::id()));
        std::fs::create_dir_all(&work).expect("the row workdir creates");
        work
    }

    /// Mints one ed25519 host key into the workdir (the independent
    /// issuer) and returns its recorded fingerprint — byte-verbatim
    /// from `ssh-keygen -lf`, the same `SHA256:` format the
    /// fingerprint parity contract pins (the policy's pin comparison
    /// is exact string equality by contract).
    fn mint_host_key(work: &Path) -> String {
        let key_path = work.join("hostkey");
        let minted = Command::new("ssh-keygen")
            .args(["-q", "-t", "ed25519", "-N", "", "-C", "alkssh-e2e-fixture"])
            .arg("-f")
            .arg(&key_path)
            .output()
            .expect("ssh-keygen spawns");
        assert!(
            minted.status.success(),
            "the host key mints — stderr: {}",
            String::from_utf8_lossy(&minted.stderr)
        );
        let listed = Command::new("ssh-keygen")
            .arg("-lf")
            .arg(key_path.with_extension("pub"))
            .output()
            .expect("ssh-keygen -lf spawns");
        let text = String::from_utf8_lossy(&listed.stdout).to_string();
        text.split_whitespace()
            .find(|token| token.starts_with("SHA256:"))
            .expect("the fingerprint line carries a SHA256: fingerprint")
            .to_owned()
    }

    /// The in-process port allocator for the rows' named binds (the
    /// e2e-server posture verbatim: a monotonic counter over a window
    /// below the kernel's ephemeral range, probe-verified per
    /// allocation — parallel rows never collide).
    pub async fn pick_port() -> u16 {
        static NEXT: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(2000);
        loop {
            let raw = NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            let port = 2000 + (raw % 15_000) as u16;
            match tokio::net::TcpListener::bind(("127.0.0.1", port)).await {
                Ok(listener) => {
                    drop(listener);
                    return port;
                }
                Err(_) => continue,
            }
        }
    }

    /// The spawned real `sshd` — the per-row fixture (the shape
    /// recorded in the module docs). The dial rides the
    /// decrypt-at-dial path over the in-memory vault double; the
    /// case-1 pinned policy accepts exactly the minted host key.
    pub struct SshdFixture {
        pub work: PathBuf,
        pub port: u16,
        pub host_fingerprint: String,
        pub username: String,
        child: tokio::process::Child,
    }

    impl SshdFixture {
        /// A fixture dialing publickey-only with TCP forwarding
        /// enabled (`AllowTcpForwarding yes`) — the op rows' server.
        pub async fn spawn(row: &str) -> Result<Self, String> {
            Self::spawn_with(row, true).await
        }

        /// A fixture with an explicit `AllowTcpForwarding` knob (the
        /// server-refused direct-tcpip row's `no` posture).
        pub async fn spawn_with(row: &str, forwarding: bool) -> Result<Self, String> {
            let Some(server) = sshd_binary() else {
                return Err("no sshd binary in the environment".to_owned());
            };
            if !keygen_present() {
                return Err("ssh-keygen is unavailable".to_owned());
            }

            let work = workdir(row);
            let host_fingerprint = mint_host_key(&work);

            let authorized = work.join("authorized_keys");
            std::fs::write(
                &authorized,
                format!("{} alkssh-e2e-user\n", fixture("alpha").blob),
            )
            .expect("the authorized_keys file writes");

            let port = pick_port().await;
            let config_path = work.join("sshd_config");
            let log_path = work.join("sshd.log");
            std::fs::write(
                &config_path,
                format!(
                    "Port {port}\n\
                     ListenAddress 127.0.0.1\n\
                     HostKey {host}\n\
                     AuthorizedKeysFile {authorized}\n\
                     PubkeyAuthentication yes\n\
                     PasswordAuthentication no\n\
                     KbdInteractiveAuthentication no\n\
                     PermitRootLogin prohibit-password\n\
                     UsePAM no\n\
                     UseDNS no\n\
                     StrictModes no\n\
                     PidFile none\n\
                     LogLevel INFO\n\
                     AllowTcpForwarding {forwarding}\n\
                     AllowAgentForwarding no\n\
                     X11Forwarding no\n",
                    host = work.join("hostkey").display(),
                    authorized = authorized.display(),
                ),
            )
            .expect("the sshd_config writes");

            let log = std::fs::File::create(&log_path).expect("the sshd log file opens");
            let mut child = tokio::process::Command::new(&server)
                .args(["-D", "-e", "-f"])
                .arg(&config_path)
                .kill_on_drop(true)
                .stderr(log)
                .spawn()
                .map_err(|error| format!("sshd refuses to spawn ({error})"))?;

            let username = Command::new("id")
                .arg("-un")
                .output()
                .ok()
                .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned())
                .filter(|user| !user.is_empty())
                .ok_or_else(|| "the connecting user cannot be resolved (id -un)".to_owned())?;

            let readiness = tokio::time::timeout(Duration::from_secs(15), async {
                loop {
                    if tokio::net::TcpStream::connect(("127.0.0.1", port))
                        .await
                        .is_ok()
                    {
                        return Ok(());
                    }
                    match child.try_wait() {
                        Ok(Some(status)) => {
                            return Err(format!(
                                "sshd exits at startup (status {status}): {}",
                                log_tail(&log_path)
                            ));
                        }
                        Ok(None) => {}
                        Err(error) => {
                            return Err(format!(
                                "sshd cannot be polled ({error}): {}",
                                log_tail(&log_path)
                            ));
                        }
                    }
                    tokio::time::sleep(Duration::from_millis(50)).await;
                }
            })
            .await;
            match readiness {
                Ok(Ok(())) => {}
                Ok(Err(reason)) => {
                    let _ = child.kill().await;
                    return Err(reason);
                }
                Err(_) => {
                    let _ = child.kill().await;
                    return Err(format!(
                        "sshd never opened its listen port: {}",
                        log_tail(&log_path)
                    ));
                }
            }

            Ok(Self {
                work,
                port,
                host_fingerprint,
                username,
                child,
            })
        }

        /// The dialed address (the dial config's `host:port` form).
        pub fn address(&self) -> String {
            format!("127.0.0.1:{}", self.port)
        }

        /// The case-1 pinned policy — accepts exactly this fixture's
        /// minted host fingerprint at `127.0.0.1` (the pin posture
        /// consumed against a real presented key). A `Some` pin
        /// substitutes another fixture's fingerprint (the mismatch
        /// row's wrong-pin posture).
        pub fn policy(&self, wrong_pin: Option<&'static str>) -> Arc<MemHostKeyPolicy> {
            let policy = Arc::new(MemHostKeyPolicy::new());
            match wrong_pin {
                Some(pin) => policy.pin("127.0.0.1", pin),
                None => policy.pin("127.0.0.1", &self.host_fingerprint),
            }
            policy
        }

        /// The authenticated dial over the fixture under the given
        /// policy — the decrypt-at-dial path (fixture `alpha` sealed
        /// into the in-memory vault, decrypted and loaded exactly at
        /// the dial), deadline-bounded.
        pub async fn dial_from(
            &self,
            policy: Arc<MemHostKeyPolicy>,
        ) -> Result<SshConnection, DialError> {
            let vault = TestVault::unlocked();
            let source = Arc::new(VaultKeySource::new(
                vault.handle(),
                Some(vault.seal(fixture("alpha"))),
            ));
            let config = DialConfig::new(
                self.address(),
                self.username.clone(),
                source,
                Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
            );
            tokio::time::timeout(ROW_DEADLINE, dial(config))
                .await
                .expect("the dial completes within the row deadline")
        }

        /// The dial the op rows use — pinned to the fixture's own
        /// key, so the authenticated connection is the expected `Ok`.
        pub async fn dial_default(&self) -> SshConnection {
            self.dial_from(self.policy(None))
                .await
                .expect("the pinned dial succeeds against the real server")
        }

        /// Stops the spawned server and removes the workdir (best
        /// effort; a panicking row leaves both to the OS).
        pub async fn stop(mut self) {
            let _ = self.child.kill().await;
            let _ = self.child.wait().await;
            let _ = std::fs::remove_dir_all(&self.work);
        }
    }

    /// The server's captured `-e` diagnostics — the skip reason's raw
    /// input, tail-truncated (best effort; a missing log is no text).
    fn log_tail(log: &Path) -> String {
        let Ok(mut file) = std::fs::File::open(log) else {
            return String::new();
        };
        let mut text = String::new();
        if file.read_to_string(&mut text).is_err() {
            return String::new();
        }
        let from = text.len().saturating_sub(512);
        text[from..]
            .lines()
            .rev()
            .take(4)
            .collect::<Vec<_>>()
            .join(" | ")
    }

    /// A tiny TCP echo server — the direct-tcpip row's real `-L`
    /// target (the bytes round trip through the server-side bridge
    /// and this server, exactly the `-L` topology).
    pub async fn spawn_echo_server() -> u16 {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the echo listener binds");
        let port = listener.local_addr().expect("local addr").port();
        tokio::spawn(async move {
            loop {
                let Ok((mut socket, _)) = listener.accept().await else {
                    return;
                };
                tokio::spawn(async move {
                    let (mut read, mut write) = socket.split();
                    let _ = tokio::io::copy(&mut read, &mut write).await;
                });
            }
        });
        port
    }

    /// The bounded byte round trip a data leg runs: the request in,
    /// the half-close (the wire EOF), the echo back, the end.
    pub async fn round_trip<S>(mut socket: S, request: &[u8]) -> Vec<u8>
    where
        S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
    {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        tokio::time::timeout(ROW_DEADLINE, async move {
            socket.write_all(request).await.expect("the request writes");
            socket.shutdown().await.expect("the half-close writes");
            let mut responded = Vec::new();
            socket
                .read_to_end(&mut responded)
                .await
                .expect("the echo reads to the end");
            responded
        })
        .await
        .expect("the data leg completes within the row deadline")
    }
}

/// The exec row (the `session` build): the family-typed exec through
/// a real server — stdout rides `data`, stderr rides
/// `extended_data(1)`, the exit status rides in-band last, and the
/// server's reaping reports the program's real exit code (a nonzero
/// exit proves the parity is the server's, not a canned 0).
#[cfg(all(unix, feature = "session"))]
mod exec_row {
    use std::time::Duration;

    use alktty::ControlMessage;
    use alktty::wire::{ChunkReader, STREAM_CTRL_OUT, STREAM_STDERR, STREAM_STDOUT};

    use super::support;

    /// Drains the session's chunk plane until the exit chunk —
    /// returns (stdout, stderr, the exit code in-band, if any). The
    /// deadline bounds a wrong row instead of hanging it.
    async fn read_to_exit(mut session: alkssh::RemoteSession) -> (Vec<u8>, Vec<u8>, Option<i32>) {
        let mut stdout = Vec::new();
        let mut stderr = Vec::new();
        let mut exit = None;
        let mut reader = ChunkReader::new(&mut session.output);
        let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
        loop {
            let chunk = match tokio::time::timeout_at(deadline, reader.read_chunk()).await {
                Ok(Ok(chunk)) => chunk,
                _ => break,
            };
            match chunk.stream_type {
                STREAM_STDOUT => stdout.extend_from_slice(&chunk.bytes),
                STREAM_STDERR => stderr.extend_from_slice(&chunk.bytes),
                STREAM_CTRL_OUT => {
                    if let Ok(ControlMessage::Exit { code }) =
                        ControlMessage::from_slice(&chunk.bytes)
                    {
                        exit = Some(code);
                    }
                    break;
                }
                _ => {}
            }
        }
        (stdout, stderr, exit)
    }

    /// - *Contract stamp:* `client.md` §op rows (row 1, the exec
    ///   parity) + §The dial surface — the wave-4 double-driven exec
    ///   row re-landed against a real `sshd`: the family-typed op
    ///   (`exec()`, the command verbatim) through the real server's
    ///   own session reaping — stdout separated from stderr, the
    ///   real exit status reported in-band last (`exit-status(3)`,
    ///   parity — not a canned 0), and the case-1 pinned policy's
    ///   accept decision on record against the really-presented key.
    #[tokio::test]
    async fn real_sshd_exec_pumps_streams_and_reports_the_exit_status() {
        let fixture = match support::SshdFixture::spawn("exec").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("exec row", &reason);
                return;
            }
        };

        let policy = fixture.policy(None);
        let connection = fixture
            .dial_from(policy.clone())
            .await
            .expect("the pinned dial succeeds against the real server");

        let session = tokio::time::timeout(
            support::ROW_DEADLINE,
            connection.exec("printf out-ok && printf err-bad 1>&2 && exit 3"),
        )
        .await
        .expect("the exec opens within the row deadline")
        .expect("the exec opens against the real server");

        let (stdout, stderr, exit) = read_to_exit(session).await;
        assert_eq!(stdout, b"out-ok", "stdout rode the data stream");
        assert_eq!(stderr, b"err-bad", "stderr rode extended_data(1)");
        assert_eq!(
            exit,
            Some(3),
            "the real server's reaping reports the program's exit status — parity, not a canned code"
        );
        assert!(
            !connection.is_closed(),
            "the connection outlives a finished exec"
        );

        let decisions = policy.decisions();
        assert_eq!(
            decisions.len(),
            1,
            "the dial called the policy exactly once — the real presented key"
        );
        let decision = &decisions[0];
        assert!(decision.accepted, "the pinned accept is the dial's verdict");
        assert_eq!(decision.presented.host, "127.0.0.1");
        assert_eq!(decision.presented.port, fixture.port);
        assert_eq!(
            decision.presented.fingerprint, fixture.host_fingerprint,
            "the presented fingerprint is the minted host key's recorded one"
        );

        connection.close().await;
        fixture.stop().await;
    }
}

/// The direct-tcpip row (the `tunnel` build): the `-L`-shaped dial
/// through a real server's forwarding — a real TCP connection to a
/// live loopback echo server, bytes end to end.
#[cfg(all(unix, feature = "tunnel"))]
mod direct_tcpip_row {
    use super::support;

    /// - *Contract stamp:* `client.md` §op rows (row 2, the
    ///   direct-tcpip dial) — the wave-4 double-driven `-L` row
    ///   re-landed against a real server: `tunnel_direct` opens the
    ///   channel on the real wire, the plain byte duplex pumps, and
    ///   a **live** TCP listener on the far side answers
    ///   byte-for-byte (the server's own connect is exercised, not a
    ///   canned echo).
    #[tokio::test]
    async fn real_sshd_direct_tcpip_round_trips_a_live_tcp_connection() {
        let fixture = match support::SshdFixture::spawn("direct-tcpip").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("direct-tcpip row", &reason);
                return;
            }
        };

        let connection = fixture.dial_default().await;
        let target_port = support::spawn_echo_server().await;

        let channel = tokio::time::timeout(
            support::ROW_DEADLINE,
            connection.tunnel_direct("127.0.0.1", u32::from(target_port)),
        )
        .await
        .expect("the tunnel op completes within the row deadline")
        .expect("the real server serves the direct-tcpip open");

        let responded = support::round_trip(channel, b"ping-alkssh-e2e\n").await;
        assert_eq!(
            responded, b"ping-alkssh-e2e\n",
            "the bytes cross the real server's forwarding end to end"
        );
        assert!(!connection.is_closed());

        connection.close().await;
        fixture.stop().await;
    }
}

/// The `-R` rows (the `tunnel` build): the port-0 mint against a real
/// server's actually-bound reply, the server-initiated
/// `forwarded-tcpip` open served through the acceptor registry (the
/// reverse-runner consumer pattern end to end), and the cancel form
/// tearing the real listen down.
#[cfg(all(unix, feature = "tunnel"))]
mod reverse_rows {
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::{ForwardedTcpip, ReverseEndpoint, TcpipForwardedAcceptor};
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    use super::support;

    /// The acceptor double — records each open's payload fields and
    /// pumps the duplex as an echo (the assembly's pump shape).
    #[derive(Clone)]
    struct AcceptorDouble {
        opens: tokio::sync::mpsc::UnboundedSender<(String, u32, String, u32)>,
    }

    #[async_trait::async_trait]
    impl TcpipForwardedAcceptor for AcceptorDouble {
        async fn accept(&self, open: ForwardedTcpip) {
            let mut stream = open.stream;
            tokio::spawn(async move {
                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;
                            }
                        }
                    }
                }
            });
            let _ = self.opens.send((
                open.connected_address,
                open.connected_port,
                open.originator_address,
                open.originator_port,
            ));
        }
    }

    /// - *Contract stamp:* `client.md` §op rows (row 4, the `-R`
    ///   listen + ingest) + `reverse-runner.md` (the consumer
    ///   pattern) — the wave-4 double-driven row re-landed against a
    ///   real server: the port-0 mint returns the **actually-bound**
    ///   port (a real server's reply, verified by dialing it), the
    ///   server-initiated `forwarded-tcpip` open arrives into the
    ///   installed acceptor, and the end-to-end consumer pattern
    ///   pumps: an external dialer → sshd's listen → the arrival →
    ///   the acceptor's duplex → the echo back.
    #[tokio::test]
    async fn real_sshd_port_zero_mint_serves_the_accepted_arrival_and_the_data_leg() {
        let fixture = match support::SshdFixture::spawn("reverse-mint").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("-R mint row", &reason);
                return;
            }
        };

        let connection = fixture.dial_default().await;
        let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
        connection
            .acceptors()
            .install_tcpip(Arc::new(AcceptorDouble { opens: opens_tx }))
            .expect("the acceptor slot installs");

        let listener = tokio::time::timeout(
            support::ROW_DEADLINE,
            connection.reverse_listen("127.0.0.1", 0),
        )
        .await
        .expect("the listen completes within the row deadline")
        .expect("the real server grants the forward");

        let bound = match listener.endpoint() {
            ReverseEndpoint::Tcp { address, port } => {
                assert_eq!(
                    address, "127.0.0.1",
                    "the wire address is the requested one"
                );
                *port
            }
            other => panic!("the tcp endpoint is the mint's shape, got {other:?}"),
        };
        assert!(
            bound != 0,
            "the mint surfaces the peer-picked port, never the requested 0"
        );

        let mut dialer = tokio::time::timeout(
            support::ROW_DEADLINE,
            tokio::net::TcpStream::connect((
                "127.0.0.1",
                u16::try_from(bound).expect("the bound port fits u16"),
            )),
        )
        .await
        .expect("the dialer connects within the row deadline")
        .expect("the minted port is actually bound — a real server's reply, dialable locally");

        let (connected_address, connected_port, originator_address, originator_port) =
            tokio::time::timeout(Duration::from_secs(15), opens_rx.recv())
                .await
                .expect("the arrival lands in time")
                .expect("the acceptor sender lives");
        assert_eq!(connected_address, "127.0.0.1");
        assert_eq!(
            connected_port, bound,
            "the arrival echoes the listen request"
        );
        assert_eq!(originator_address, "127.0.0.1");
        assert!(
            originator_port != 0,
            "the originator is the external dialer's ephemeral source port"
        );

        let responded = support::round_trip(&mut dialer, b"reverse-leg-marker\n").await;
        assert_eq!(
            responded, b"reverse-leg-marker\n",
            "the consumer pattern's data leg round trips through the real server"
        );

        assert!(!connection.is_closed());
        drop(listener);
        connection.close().await;
        fixture.stop().await;
    }

    /// - *Contract stamp:* `client.md` §Lifecycle primitives (the
    ///   cancel form) — the wave-4 double-driven cancel row
    ///   re-landed against a real server: an explicit-port listen is
    ///   actually bound (dialable), the listener handle's `cancel()`
    ///   tears the **real** listen down (a later dial is refused), a
    ///   stateless cancel of the unknown listener is the wire's
    ///   `ListenRefused`, and the connection stays usable.
    #[tokio::test]
    async fn real_sshd_cancel_tears_the_listen_down_and_unknown_cancels_refuse() {
        let fixture = match support::SshdFixture::spawn("reverse-cancel").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("-R cancel row", &reason);
                return;
            }
        };

        let connection = fixture.dial_default().await;
        let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel();
        connection
            .acceptors()
            .install_tcpip(Arc::new(AcceptorDouble { opens: opens_tx }))
            .expect("the acceptor slot installs");

        let port = support::pick_port().await;
        let listener = tokio::time::timeout(
            support::ROW_DEADLINE,
            connection.reverse_listen("127.0.0.1", u32::from(port)),
        )
        .await
        .expect("the listen completes within the row deadline")
        .expect("the real server grants the forward");
        assert_eq!(
            listener.endpoint().bound_port(),
            Some(u32::from(port)),
            "an explicit-port listen surfaces the requested port (the wire's mint data is absent)"
        );

        let _bound_conn = tokio::time::timeout(
            support::ROW_DEADLINE,
            tokio::net::TcpStream::connect(("127.0.0.1", port)),
        )
        .await
        .expect("the probe connects within the row deadline")
        .expect("the explicit port is actually bound before the cancel");

        tokio::time::timeout(support::ROW_DEADLINE, listener.cancel())
            .await
            .expect("the cancel completes within the row deadline")
            .expect("the real listen tears down");

        let mut torn_down = false;
        for _ in 0..40 {
            match tokio::net::TcpStream::connect(("127.0.0.1", port)).await {
                Ok(_) => tokio::time::sleep(Duration::from_millis(100)).await,
                Err(_) => {
                    torn_down = true;
                    break;
                }
            }
        }
        assert!(
            torn_down,
            "the cancel frees the real bound port — connects are refused afterwards"
        );

        let again = connection
            .reverse_listen_cancel("127.0.0.1", u32::from(port))
            .await;
        assert!(
            matches!(again, Err(alkssh::OpError::ListenRefused)),
            "the real server refuses the unknown listener, got {again:?}"
        );
        assert!(
            !connection.is_closed(),
            "the connection stays usable after every cancel class"
        );

        connection.close().await;
        fixture.stop().await;
    }
}

/// The decline rows — the sftp placeholder (every build) and the
/// host-key policy mismatch (every build): both dial a real server
/// and pin typed declines, feature-free.
#[cfg(unix)]
mod declines {
    use std::sync::Arc;

    use alkssh::{DialError, OpError};

    use super::common::fixture as key_fixture;
    use super::support;

    /// - *Contract stamp:* OQ-13 + `client.md` §op rows (row 5) —
    ///   the sftp placeholder declines in **every** build against a
    ///   real authenticated connection: typed decline before any
    ///   wire op (the real server observes no subsystem request),
    ///   connection intact.
    #[tokio::test]
    async fn real_connection_sftp_placeholder_declines_with_the_connection_intact() {
        let fixture = match support::SshdFixture::spawn("sftp-decline").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("sftp decline row", &reason);
                return;
            }
        };

        let connection = fixture.dial_default().await;
        let err = tokio::time::timeout(support::ROW_DEADLINE, connection.sftp())
            .await
            .expect("the op completes within the row deadline")
            .expect_err("the OQ-13 placeholder declines against the real connection");
        assert!(matches!(err, OpError::SftpNotServed), "got {err:?}");
        assert!(
            !connection.is_closed(),
            "the pre-wire decline touches no wire state — the connection stays up"
        );
        connection.close().await;
        fixture.stop().await;
    }

    /// - *Contract stamp:* `client.md` §The dial surface (the policy
    ///   call at `check_server_key`, OQ-05) + Host-key §case 1 — the
    ///   pin posture consumed against a real presented key: a
    ///   mismatch hard-declines the dial (`DialError::
    ///   HostKeyRejected`, no connection), the policy's decision log
    ///   records the **real** fingerprint as rejected, and a
    ///   correctly-pinned dial to the same server succeeds — the
    ///   decline is the policy's decision, not the server's.
    #[tokio::test]
    async fn real_presented_key_mismatch_hard_declines_and_the_right_pin_dials() {
        let fixture = match support::SshdFixture::spawn("host-key-mismatch").await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("host-key mismatch row", &reason);
                return;
            }
        };

        let wrong = fixture.policy(Some(key_fixture("delta").fingerprint));
        let declined = fixture
            .dial_from(Arc::clone(&wrong))
            .await
            .expect_err("the mismatch hard-declines the dial");
        assert!(
            matches!(declined, DialError::HostKeyRejected),
            "got {declined:?}"
        );

        let decisions = wrong.decisions();
        assert_eq!(
            decisions.len(),
            1,
            "the dial called the policy exactly once — the real presented key"
        );
        let decision = &decisions[0];
        assert!(
            !decision.accepted,
            "the pinned mismatch is the reject verdict"
        );
        assert_eq!(decision.presented.host, "127.0.0.1");
        assert_eq!(decision.presented.port, fixture.port);
        assert_eq!(
            decision.presented.fingerprint, fixture.host_fingerprint,
            "the presented fingerprint is the real server's minted host key"
        );

        let connection = fixture.dial_default().await;
        assert!(
            !connection.is_closed(),
            "the correctly-pinned dial to the same server succeeds"
        );
        connection.close().await;
        fixture.stop().await;
    }
}

/// The server-refused direct-tcpip row (the `tunnel` **and**
/// `session` builds — the all-features posture): a real server with
/// `AllowTcpForwarding no` refuses the typed open, and the
/// connection stays usable (a following exec is served). The session
/// gate is the stays-usable proof's vehicle (recorded: under
/// tunnel-only builds the row is absent; the fake-double mirror,
/// `tests/client_ops.rs`, carries the tunnel-only stays-usable rows).
#[cfg(all(unix, feature = "tunnel", feature = "session"))]
mod unfeatured_direct_decline {
    use std::time::Duration;

    use alkssh::{OpError, OpenDeclineReason};

    use super::support;

    /// - *Contract stamp:* `client.md` §op rows (row 2's declined
    ///   mirror) + §Correctness facts (a decline is channel-scoped,
    ///   auth intact, connection usable) — the wave-4 typed-open
    ///   decline row re-landed against a real server's own refusal:
    ///   `tunnel_direct` against `AllowTcpForwarding no` →
    ///   `OpenDeclined(Prohibited)` (the `administratively
    ///   prohibited` class), and a following exec is served — the
    ///   connection outlives the decline end to end.
    #[tokio::test]
    async fn real_server_refused_direct_tcpip_declines_and_the_connection_stays_usable() {
        let fixture = match support::SshdFixture::spawn_with("direct-decline", false).await {
            Ok(fixture) => fixture,
            Err(reason) => {
                support::skip("direct-tcpip decline row", &reason);
                return;
            }
        };

        let connection = fixture.dial_default().await;

        let declined = tokio::time::timeout(
            support::ROW_DEADLINE,
            connection.tunnel_direct("127.0.0.1", 443),
        )
        .await
        .expect("the tunnel op completes within the row deadline")
        .expect_err("the server that forbids forwarding refuses the open");
        assert!(
            matches!(
                declined,
                OpError::OpenDeclined(OpenDeclineReason::Prohibited)
            ),
            "the administratively-prohibited class mirrors through, got {declined:?}"
        );
        assert!(
            !connection.is_closed(),
            "the decline is channel-scoped — the connection survives it"
        );

        let mut session =
            tokio::time::timeout(support::ROW_DEADLINE, connection.exec("echo still-usable"))
                .await
                .expect("the exec opens within the row deadline")
                .expect("the post-decline exec is served — the connection stays usable");

        let mut stdout = Vec::new();
        let mut reader = alktty::wire::ChunkReader::new(&mut session.output);
        let deadline = tokio::time::Instant::now() + Duration::from_secs(15);
        loop {
            match tokio::time::timeout_at(deadline, reader.read_chunk()).await {
                Ok(Ok(chunk)) if chunk.stream_type == alktty::wire::STREAM_STDOUT => {
                    stdout.extend_from_slice(&chunk.bytes);
                }
                Ok(Ok(chunk)) if chunk.stream_type == alktty::wire::STREAM_CTRL_OUT => {
                    break;
                }
                _ => break,
            }
        }
        assert_eq!(
            stdout, b"still-usable\n",
            "the exec after the decline answers — auth intact, connection usable"
        );

        connection.close().await;
        fixture.stop().await;
    }
}