//! The server-direction real-tool e2e (`w5-e2e-server`) — the POC-3
//! gates re-landed against the shipped crate: real vanilla processes
//! (`ssh` / `git` / `curl` / `sftp`) against the adapter mounted on a
//! real TCP listener over the established in-memory doubles (the
//! wave-3 smoke composition, `tests/common/`). The row set is the
//! full server direction: OpenSSH pty/shell and pipe exec, `-L`,
//! `-R`, `-D` (the `direct-tcpip` leg), the decline-observability
//! rows (unfeatured declines from real attempts, the sftp subsystem
//! decline, the unknown-repo collapsed generic denial), and git
//! clone **and** push through the git mapping.
//!
//! **Placement decision (recorded, the task's acceptance criterion
//! 1):** a dedicated `tests/e2e_server.rs` binary reusing
//! `tests/common` — the wave-3/4 mapping binaries keep their
//! single-writer posture for `w5-decline-matrix`; this binary owns
//! only its own rows and its in-file fixture helpers (the trimmed
//! leg-1/leg-2 fabric double for the tunnel rows, the support
//! helpers). Rows arrive under the established check-gated
//! **skipped-with-reason** posture (tool absent / version short —
//! never silently green) and under a wall-clock deadline — a decline
//! must surface a clean diagnostic to the real client, never a hang
//! (POC-1 §5 posture). Rows carry `Contract stamp:` doc markers
//! against `mapping.md` / the POC-3 findings (the greppable
//! `tests/common` discipline).

mod common;

/// The per-binary shared helpers: tool gates, the fixture-key /
/// option builders, the registered identity provider, the ephemeral
/// port picker, and the real-TCP mount every row composes.
#[cfg(unix)]
mod support {
    #![allow(dead_code)]

    use std::path::{Path, PathBuf};
    use std::process::Command;
    use std::sync::Arc;

    use alkssh::{ServeConfig, SshAdapter};

    use super::common::{MemIdentityProvider, fixture, identity};

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

    /// True when spawning the probe succeeds at all (exit codes are
    /// the probe's own business — `ssh -V` writes to stderr, `sftp -h`
    /// exits 1 with usage).
    pub fn tool_present(binary: &str, probe: &[&str]) -> bool {
        Command::new(binary).args(probe).output().is_ok()
    }

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

    /// The per-row throwaway workdir + the fixture key written into
    /// it (0600 — the real client's required file mode).
    pub fn write_key(row: &str) -> PathBuf {
        let work =
            std::env::temp_dir().join(format!("alkssh-e2e-server-{row}-{}", std::process::id()));
        std::fs::create_dir_all(&work).expect("the row 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");
        key_path
    }

    /// A row's cleanup (best-effort on the success path; a panicking
    /// row leaves the temp dir for the OS).
    pub fn cleanup(work: &Path) {
        let _ = std::fs::remove_dir_all(work);
    }

    /// The decline-event observation for the e2e rows: poll the
    /// capture until a matching event is on record — the row deadline
    /// bounds the wait (not the capture's fixed window), so a slow CI
    /// host never turns an honest decline into a flake.
    pub async fn decline_event(want: &[(&str, &str)]) -> super::common::event_capture::Captured {
        let deadline = std::time::Instant::now() + ROW_DEADLINE;
        loop {
            if let Some(event) = super::common::event_capture::captured_with(want) {
                return event;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "no decline event matching {want:?} within the row deadline"
            );
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
    }

    /// The OpenSSH client option set every row shares (batch mode, no
    /// known-hosts persistence, bounded connect).
    pub fn ssh_options() -> Vec<String> {
        vec![
            "-o".to_owned(),
            "BatchMode=yes".to_owned(),
            "-o".to_owned(),
            "IdentitiesOnly=yes".to_owned(),
            "-o".to_owned(),
            "StrictHostKeyChecking=no".to_owned(),
            "-o".to_owned(),
            "UserKnownHostsFile=/dev/null".to_owned(),
            "-o".to_owned(),
            "ConnectTimeout=10".to_owned(),
        ]
    }

    /// The `GIT_SSH_COMMAND` transport line for the git rows (the
    /// same option set, one string).
    pub fn git_ssh_command(key: &Path) -> String {
        format!(
            "ssh -i {} -o BatchMode=yes -o IdentitiesOnly=yes \
             -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \
             -o ConnectTimeout=10",
            key.display()
        )
    }

    /// The git 2.43+ availability gate (the wave-3 smoke's version
    /// gate, shared by every git row).
    pub fn git_available() -> (bool, String) {
        let output = Command::new("git").arg("--version").output();
        let Ok(output) = output else {
            return (false, "git is unavailable".to_owned());
        };
        let text = String::from_utf8_lossy(&output.stdout).to_string();
        let version = text
            .split_whitespace()
            .last()
            .unwrap_or_default()
            .split('.')
            .filter_map(|c| c.parse::<u32>().ok())
            .collect::<Vec<u32>>();
        if version.first() == Some(&2) && version.get(1).is_some_and(|minor| *minor >= 43) {
            (true, text.trim().to_owned())
        } else {
            (false, format!("git {text:?} is older than 2.43.0"))
        }
    }

    /// The registered identity provider every row serves against:
    /// fixture `alpha`'s fingerprint resolves to the row identity
    /// (the ADR-004 decision 1 resolution shape).
    pub fn provider(identity_id: &str) -> Arc<MemIdentityProvider> {
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity(identity_id, &["ssh"]),
        );
        provider
    }

    /// The in-process port allocator for the rows' named binds
    /// (`-L`/`-D` client-side listeners, the `-R` fabric's establisher
    /// bind): a monotonic counter over a window **below** the kernel's
    /// ephemeral port range, so parallel rows can never collect the
    /// same port and bind(0) allocations can never steal one — the
    /// probe-bind re-verifies availability (a previous run's TIME_WAIT
    /// tuples) and skips taken ports.
    pub async fn pick_local_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,
            }
        }
    }

    /// Mounts the door over a real TCP loopback listener over the
    /// in-memory doubles (the wave-3 smoke's accept-loop shape): the
    /// adapter constructs over the serve config, binds an ephemeral
    /// port, and serves each accepted connection to completion on a
    /// spawned task. The returned port is what the real tools dial.
    pub async fn mount(serve: ServeConfig) -> (u16, tokio::task::JoinHandle<()>) {
        let adapter = Arc::new(
            SshAdapter::new(serve)
                .await
                .expect("the e2e adapter constructs"),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the e2e listener binds");
        let 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;
                });
            }
        });
        (port, accept_task)
    }
}

/// The git rows — `git` build: clone (the wave-3 smoke re-landed) and
/// push (the POC-3 stage-B receive-pack gate), through the git
/// mapping over the hand-rolled recording quartet.
#[cfg(unix)]
#[cfg(feature = "git")]
mod git_rows {
    use std::process::Output;
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::Limits;

    use super::common::git_doubles as gd;
    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    /// The unborn-HEAD advertisement every row serves (the wave-3
    /// smoke's repo record).
    fn unborn_demo(doubles: &Arc<gd::MemGit>) {
        doubles.install(gd::RepoSetup::private_granted(
            "owner/demo",
            "git-rider",
            vec![gd::RefLine {
                name: "HEAD".into(),
                oid: None,
                peeled: None,
                symref_target: Some("refs/heads/main".into()),
            }],
        ));
    }

    async fn git_mount(doubles: &Arc<gd::MemGit>) -> (u16, tokio::task::JoinHandle<()>) {
        let provider = support::provider("git-rider");
        let mut serve = harness_serve_config(
            provider,
            &[super::common::fixture("bravo")],
            Limits::default(),
        );
        serve.git_seam = Some(gd::git_seam(doubles));
        support::mount(serve).await
    }

    async fn run_git(args: &[&str]) -> Output {
        tokio::process::Command::new("git")
            .args(args)
            .output()
            .await
            .expect("git spawns")
    }

    async fn run_git_ssh(args: &[&str], ssh: &str) -> Output {
        tokio::process::Command::new("git")
            .args(args)
            .env("GIT_SSH_COMMAND", ssh)
            .output()
            .await
            .expect("git spawns")
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (the git exec
    ///   row — `git clone ssh://…`; POC-3 §A.1 re-landed against the
    ///   shipped crate).
    #[tokio::test]
    async fn real_git_clones_the_empty_repo_through_the_door() {
        let (git_ok, git_reason) = support::git_available();
        if !git_ok {
            support::skip("git clone row", &git_reason);
            return;
        }
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("git clone row", "ssh is unavailable");
            return;
        }

        let key = support::write_key("git-clone");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let doubles = Arc::new(gd::MemGit::new());
        unborn_demo(&doubles);
        let (port, accept_task) = git_mount(&doubles).await;

        let ssh = support::git_ssh_command(&key);
        let target = work.join("clone");
        let outcome = run_git_ssh(
            &[
                "clone",
                &format!("ssh://git-rider@127.0.0.1:{port}/owner/demo.git"),
                target.to_str().expect("the clone target path"),
            ],
            &ssh,
        )
        .await;

        accept_task.abort();

        let stderr = String::from_utf8_lossy(&outcome.stderr).to_string();
        assert!(
            outcome.status.success(),
            "vanilla git clone (2.43+) through the door succeeds — stderr: {stderr}"
        );
        assert!(
            target.join(".git").is_dir(),
            "the clone materializes a repository (the empty-repo clone: unborn HEAD)"
        );
        assert!(
            stderr.contains("empty repository"),
            "the unborn-HEAD advertisement is what the clone consumed — stderr: {stderr}"
        );
        assert!(
            doubles.resolved_ids().iter().any(|id| id == "owner/demo"),
            "the resolved id is the normalized one (the .git suffix rode off)"
        );
        support::cleanup(&work);
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (the git exec
    ///   row — `git push ssh://…` = receive-pack through the real
    ///   client; POC-3 §A stage-B gate re-landed, the wave-3 smoke's
    ///   clone-only coverage completed).
    #[tokio::test]
    async fn real_git_push_creates_the_ref_through_the_door() {
        let (git_ok, git_reason) = support::git_available();
        if !git_ok {
            support::skip("git push row", &git_reason);
            return;
        }
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("git push row", "ssh is unavailable");
            return;
        }

        let key = support::write_key("git-push");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let doubles = Arc::new(gd::MemGit::new());
        unborn_demo(&doubles);
        let (port, accept_task) = git_mount(&doubles).await;

        let repo = work.join("local");
        let init = run_git(&[
            "-c",
            "init.defaultBranch=main",
            "init",
            repo.to_str().expect("the local repo path"),
        ])
        .await;
        assert!(
            init.status.success(),
            "the local repo initializes — stderr: {}",
            String::from_utf8_lossy(&init.stderr)
        );
        std::fs::write(repo.join("hello.txt"), "alkssh e2e push\n")
            .expect("the commit file writes");
        let add = run_git(&["-C", repo.to_str().expect("path"), "add", "-A"]).await;
        assert!(add.status.success(), "the staged tree adds");
        let commit = run_git(&[
            "-C",
            repo.to_str().expect("path"),
            "-c",
            "user.name=e2e-rider",
            "-c",
            "user.email=e2e@alk.dev",
            "commit",
            "-m",
            "alkssh e2e push row",
        ])
        .await;
        assert!(
            commit.status.success(),
            "the commit lands — stderr: {}",
            String::from_utf8_lossy(&commit.stderr)
        );
        let head = run_git(&["-C", repo.to_str().expect("path"), "rev-parse", "HEAD"]).await;
        assert!(head.status.success(), "rev-parse HEAD");
        let head_hex = String::from_utf8_lossy(&head.stdout).trim().to_owned();

        let ssh = support::git_ssh_command(&key);
        let push = tokio::time::timeout(
            Duration::from_secs(30),
            run_git_ssh(
                &[
                    "-C",
                    repo.to_str().expect("path"),
                    "push",
                    &format!("ssh://git-rider@127.0.0.1:{port}/owner/demo.git"),
                    "main",
                ],
                &ssh,
            ),
        )
        .await;

        accept_task.abort();

        let outcome = match push {
            Ok(outcome) => outcome,
            Err(_) => panic!("the real git push row exceeded its 30 s deadline"),
        };
        let stderr = String::from_utf8_lossy(&outcome.stderr).to_string();
        assert!(
            outcome.status.success(),
            "vanilla git push (receive-pack) through the door succeeds — stderr: {stderr}"
        );

        let prepared = doubles.prepared();
        let update = prepared
            .first()
            .and_then(|pack| pack.updates.iter().find(|u| u.name == "refs/heads/main"))
            .expect("the push prepared one receive-pack carrying refs/heads/main");
        let new = update.new.expect("a branch push carries the new ref oid");
        assert_eq!(
            format!("{new}"),
            head_hex,
            "the ref the door's receive-pack applied is the pushed commit (rev parity)"
        );
        assert!(
            doubles
                .applied()
                .iter()
                .any(|applied| applied.updates.iter().any(|u| u.name == "refs/heads/main")),
            "the CAS application rides apply_updates (the wire layer's single ingestion home)"
        );
        support::cleanup(&work);
    }
}

/// The session rows — `session` build: the interactive pty/shell
/// session (the wave-3 smoke re-landed, extended to the echo marker)
/// and the plain pipe-exec row with its separate stderr.
#[cfg(unix)]
#[cfg(feature = "session")]
mod session_rows {
    use std::collections::HashMap;
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::{Limits, SessionSeam};
    use alktty::LocalTtyBackend;
    use alktty::backend::TtyBackend;

    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    /// The `local` composition for the session rows — alktty's own
    /// backend in the seam map (dev-dep unification: the `local`
    /// feature rides the test graph only).
    fn local_seam() -> SessionSeam {
        let mut backends = HashMap::new();
        backends.insert(
            "local".to_owned(),
            Arc::new(LocalTtyBackend::new()) as Arc<dyn TtyBackend>,
        );
        SessionSeam {
            backends: Arc::new(backends),
            backend: "local".to_owned(),
            ownership: None,
        }
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (the
    ///   `session` → `pty_request` + `shell_request` row; POC-3 §D.1
    ///   re-landed as a full OpenSSH interactive session).
    #[tokio::test]
    async fn real_openssh_pty_shell_session_serves_through_the_door() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("pty/shell row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("sh", &["-c", "exit 0"]) {
            support::skip("pty/shell row", "sh is unavailable");
            return;
        }

        let key = support::write_key("pty-shell");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let mut serve = harness_serve_config(
            support::provider("e2e-rider"),
            &[super::common::fixture("bravo")],
            Limits::default(),
        );
        serve.shell_argv = Some(vec!["/bin/sh".to_owned(), "-i".to_owned()]);
        serve.session_seam = Some(local_seam());
        let (port, accept_task) = support::mount(serve).await;

        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            let mut child = tokio::process::Command::new("ssh")
                .args(["-tt", "-p", &port.to_string(), "-i"])
                .arg(&key)
                .args(support::ssh_options())
                .arg("e2e-rider@127.0.0.1")
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .spawn()
                .expect("ssh spawns");
            {
                use tokio::io::AsyncWriteExt;
                let mut stdin = child.stdin.take().expect("the child stdin is piped");
                stdin
                    .write_all(b"echo alkssh-e2e-pty; exit\n")
                    .await
                    .expect("the row stdin writes");
                stdin.flush().await.expect("the row stdin flushes");
            }
            child.wait_with_output().await.expect("ssh completes")
        })
        .await;

        accept_task.abort();

        let output = match outcome {
            Ok(output) => output,
            Err(_) => panic!("the pty/shell row exceeded its 30 s deadline"),
        };
        let stdout = String::from_utf8_lossy(&output.stdout).to_string();
        assert!(
            output.status.success(),
            "vanilla OpenSSH's pty+shell session through the door exits 0 \
             (the exit chunk derived exit-status(0) before close) — stdout: {stdout:?}"
        );
        assert!(
            stdout.contains("alkssh-e2e-pty"),
            "the keystrokes and the shell's echo rode the real PTY round trip — stdout: {stdout:?}"
        );
        support::cleanup(&work);
    }

    /// - *Contract stamp:* mapping.md §Request-level rows (the
    ///   `exec_request` pipe row — `wrapper ++ [cmd]`, stderr rides
    ///   `extended_data(1)`; POC-3 §D.1's pipe split, end to end).
    #[tokio::test]
    async fn real_openssh_pipe_exec_separates_stdout_and_stderr_through_the_door() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("pipe exec row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("sh", &["-c", "exit 0"]) {
            support::skip("pipe exec row", "sh is unavailable");
            return;
        }

        let key = support::write_key("pipe-exec");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let mut serve = harness_serve_config(
            support::provider("e2e-rider"),
            &[super::common::fixture("bravo")],
            Limits::default(),
        );
        serve.exec_wrapper_argv = Some(vec!["/bin/sh".to_owned(), "-c".to_owned()]);
        serve.session_seam = Some(local_seam());
        let (port, accept_task) = support::mount(serve).await;

        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            tokio::process::Command::new("ssh")
                .args(["-p", &port.to_string(), "-i"])
                .arg(&key)
                .args(support::ssh_options())
                .arg("e2e-rider@127.0.0.1")
                .arg("echo alkssh-e2e-stdout; echo alkssh-e2e-stderr >&2")
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .output()
                .await
                .expect("ssh spawns and completes")
        })
        .await;

        accept_task.abort();

        let output = match outcome {
            Ok(output) => output,
            Err(_) => panic!("the pipe-exec row exceeded its 30 s deadline"),
        };
        let stdout = String::from_utf8_lossy(&output.stdout).to_string();
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();
        assert!(
            output.status.success(),
            "vanilla `ssh host \"cmd\"` through the exec wrapper exits 0 — stdout: {stdout:?} stderr: {stderr:?}"
        );
        assert!(
            stdout.contains("alkssh-e2e-stdout"),
            "the pipe row's stdout rides SSH data — stdout: {stdout:?}"
        );
        assert!(
            stderr.contains("alkssh-e2e-stderr"),
            "the pipe row's stderr rides extended_data(1) separately — stderr: {stderr:?}"
        );
        support::cleanup(&work);
    }
}

/// The tunnel rows — `tunnel` build: `-L` (the route-1 bridge), `-R`
/// (listen + arrival + data leg through the leg-2 fabric), and `-D`
/// (curl's SOCKS traffic takes the `direct-tcpip` leg — POC-3
/// correction 2's posture).
#[cfg(unix)]
#[cfg(feature = "tunnel")]
mod tunnel_rows {
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::{Limits, ServeConfig, TunnelListenSeam};

    use super::common::harness::serve_config as harness_serve_config;
    use super::common::{fixture, identity};
    use super::support;

    /// The in-file fabric fixtures — the trimmed leg-1/leg-2 fabric
    /// double (the wave-3 smoke's recorded composition, POC-3 §C.4
    /// recipe), TCP-only and happy-path only: the decline/collision
    /// machinery stays the mapping binaries'.
    #[cfg(unix)]
    mod fabric {
        #![allow(dead_code)]

        use std::sync::Arc;
        use std::sync::atomic::{AtomicU32, Ordering};

        use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
        use alkcall::channels::client::{ChannelClient, ServingConfig};
        use alkcall::channels::operations::{
            ChannelCore, ChannelOperations, Establishment, OpenEstablisher,
        };
        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::{
            OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN, Substrate, SubstrateAddr, 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 crate::common::identity;

        /// The registration settle: the fabric's openables register on
        /// a spawned task, so a row that drives a client before the
        /// registrations land races (a listen dial before
        /// `channels/tunnel/sub` is registered is refused, not
        /// retried). Every fabric-consumer waits for the named ops'
        /// registrations with a bounded poll.
        async fn settle(registry: &Arc<OperationRegistry>, ops: &[&str]) {
            for _ in 0..6000 {
                if ops.iter().all(|op| registry.registration(op).is_some()) {
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
            panic!("the fabric never registered {ops:?}");
        }

        /// The dial seam: a real-TCP dial closure (the assembly's
        /// injected `DialFn` for real; alktunnels' `local` feature
        /// stays off).
        pub(super) 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),
                    })
                })
            })
        }

        /// The leg-1 fabric (`w3-tunnel-direct`'s recorded double): an
        /// `alk/channels` connection whose accept side registers
        /// `ChannelOperations` + the `channels/tunnel/direct` openable
        /// (the injected dial seam), and whose connect side is the
        /// `Arc<ChannelClient>` a tunnel-featured serve-config holds.
        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 caller_for_hook = caller.clone();
            let dial_for_hook = Arc::clone(&dial);
            let witness_for_hook = Some(witness);

            let install_hook: InstallChannelZero =
                Arc::new(move |manager, channel0_conn, _auth| {
                    let registry = Arc::clone(&registry_for_hook);
                    let caller = caller_for_hook.clone();
                    let dial = Arc::clone(&dial_for_hook);
                    let witness = witness_for_hook.clone();
                    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(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,
                            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;
            });

            settle(&registry, &[OP_TUNNEL_DIRECT]).await;
            Arc::new(
                ChannelClient::from_connection(consumer_conn)
                    .await
                    .expect("the fabric's channels client establishes"),
            )
        }

        /// The e2e fabric double for the `-R` row: the leg-1 shape
        /// extended with the **bind-first listen half** — the
        /// establisher binds the wire-named endpoint for real (the
        /// dynamic-address producer), mints the correlation token per
        /// the ADR-008 discipline, and the accept handler dials the
        /// `tunnel/forwarded` op at the door per arrival. The door
        /// side rides the `ServingConfig.registry` the row hands
        /// alkssh through the `TunnelListenSeam` (one `Arc`).
        pub(super) struct FabricDouble {
            pub(super) client: Arc<ChannelClient>,
            pub(super) registry: Arc<OperationRegistry>,
        }

        fn real_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(),
                        },
                    })
                })
            })
        }

        fn real_accept_loop() -> BindFirstAcceptLoop {
            Arc::new(move |listener: Arc<dyn std::any::Any + Send + Sync>| {
                Box::pin(async move {
                    let tcp = listener.downcast_ref::<tokio::net::TcpListener>().ok_or(
                        TunnelEstablishError::HandlerError(
                            "the plan's listener is not the double's bind".into(),
                        ),
                    )?;
                    let (stream, peer) = tcp
                        .accept()
                        .await
                        .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?;
                    let (read, write) = stream.into_split();
                    Ok((
                        SubstrateAddr::HostPort {
                            host: peer.ip().to_string(),
                            port: peer.port(),
                        },
                        TargetHandle {
                            read: Box::new(read),
                            write: Box::new(write),
                        },
                    ))
                })
            })
        }

        fn counter_mint() -> OpenEstablisher {
            let counter = Arc::new(AtomicU32::new(1));
            Arc::new(move |input: serde_json::Value, _auth| {
                let counter = Arc::clone(&counter);
                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 next = counter.fetch_add(1, Ordering::Relaxed);
                    let token = if next == 0 { 1 } else { next };
                    let BoundListener { listener, bound } =
                        real_tcp_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)))
                })
            })
        }

        pub(super) async fn wire_fabric_with_listen(dial: DialFn) -> FabricDouble {
            let door_registry = Arc::new(OperationRegistry::new());
            let registry = Arc::new(OperationRegistry::new());
            let registry_for_hook = Arc::clone(&registry);
            let door_caller = identity("door-caller", &[TUNNEL_OPEN_SCOPE]);
            let arrival_identity = identity("fabric-arrival", &[TUNNEL_FORWARDED_SCOPE]);

            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 dial = Arc::clone(&dial);
                    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),
                        ));

                        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;
                                    let out = response.result.map_err(|e| {
                                        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 ops = ChannelOperations::with_default_policy(manager.clone());
                        ops.register_on(&registry).expect("channel ops register");
                        let direct_core = ChannelCore::new(
                            manager.clone(),
                            alkcall::channels::policy::default_policy(),
                        );
                        register_tunnel_direct_openable(
                            &direct_core,
                            &registry,
                            AuthContext::anonymous(b"alk/tunnel"),
                            dial,
                            None,
                        )
                        .expect("the direct openable registers");
                        direct_core
                            .register_openable_with_establisher(
                                alktunnels::params::tunnel_open_spec(),
                                Some(counter_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");

            settle(&registry, &[OP_TUNNEL_DIRECT, OP_TUNNEL_OPEN]).await;

            FabricDouble {
                client: Arc::new(client),
                registry: door_registry,
            }
        }
    }

    /// 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 row endpoint binds");
        let addr = listener.local_addr().expect("local addr");
        let body = "alkssh w5-e2e-server 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 {
                    use tokio::io::{AsyncReadExt, AsyncWriteExt};
                    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)
    }

    /// The curl-until-settled loop (the -L listener binds client-side;
    /// the -R fabric binds after the listen flow): a bounded settle
    /// window, the first successful body wins.
    async fn curl_until_success(url: &str, extra: &[&str]) -> std::process::Output {
        let mut attempts = 0u32;
        loop {
            let out = tokio::process::Command::new("curl")
                .args(["-s", "--max-time", "10"])
                .args(extra)
                .arg(url)
                .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(Duration::from_millis(250)).await;
        }
    }

    async fn tunnel_mount(serve: ServeConfig) -> (u16, tokio::task::JoinHandle<()>) {
        support::mount(serve).await
    }

    fn key_arg(row: &str) -> (std::path::PathBuf, std::path::PathBuf) {
        let key = support::write_key(row);
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();
        (work, key)
    }

    fn ssh_forward_args(
        port: u16,
        key: &std::path::Path,
        forward: &[&str],
    ) -> tokio::process::Command {
        let mut cmd = tokio::process::Command::new("ssh");
        cmd.arg("-N")
            .arg("-i")
            .arg(key)
            .arg("-p")
            .arg(port.to_string())
            .args(support::ssh_options())
            .args(["-o", "ExitOnForwardFailure=yes"])
            .args(forward)
            .arg("e2e-rider@127.0.0.1")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped());
        cmd
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (the
    ///   `direct-tcpip` row — `ssh -L` through the route-1 bridge;
    ///   POC-3 §C.1 re-landed with the shipped crate).
    #[tokio::test]
    async fn real_ssh_local_forward_serves_the_direct_tcpip_leg() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("-L row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("curl", &["--version"]) {
            support::skip("-L row", "curl is unavailable");
            return;
        }

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

        let (work, key) = key_arg("tunnel-l");
        let mut serve = harness_serve_config(
            support::provider("tunnel-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        serve.channel_client = Some(fabric);
        let (ssh_port, accept_task) = tunnel_mount(serve).await;

        let fwd_port = support::pick_local_port().await;
        let mut ssh_child = ssh_forward_args(
            ssh_port,
            &key,
            &[
                "-L",
                &format!(
                    "127.0.0.1:{fwd_port}:{}:{}",
                    http_addr.ip(),
                    http_addr.port()
                ),
            ],
        )
        .spawn()
        .expect("vanilla ssh spawns");

        let through = tokio::time::timeout(Duration::from_secs(30), async {
            curl_until_success(&format!("http://127.0.0.1:{fwd_port}/"), &[]).await
        })
        .await;

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

        let out = through.expect("the -L row resolves in time");
        let stdout = String::from_utf8_lossy(&out.stdout).to_string();
        assert!(
            stdout.trim_end() == body.trim_end(),
            "vanilla ssh -L + curl through the door → fabric → endpoint returns the body — got {stdout:?}"
        );
        support::cleanup(&work);
    }

    /// - *Contract stamp:* mapping.md §Global-request rows (the `-R`
    ///   listen flow — `ssh -R`'s listen, the fabric-bound port, the
    ///   server-initiated `forwarded-tcpip` arrival, the client-dialed
    ///   data leg; the `w3-tunnel-listen` flow re-landed against a
    ///   real OpenSSH client).
    #[tokio::test]
    async fn real_ssh_remote_forward_serves_the_arrival_and_data_legs() {
        use alktunnels::params::Substrate;

        if !support::tool_present("ssh", &["-V"]) {
            support::skip("-R row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("curl", &["--version"]) {
            support::skip("-R row", "curl is unavailable");
            return;
        }

        let (http_addr, body) = spawn_http_endpoint().await;
        let fail_dial: alktunnels::producer::DialFn =
            Arc::new(move |_substrate: Substrate, target: &str| {
                let target = target.to_owned();
                Box::pin(async move {
                    Err::<
                        alktunnels::producer::TargetHandle,
                        alktunnels::producer::TunnelEstablishError,
                    >(alktunnels::producer::TunnelEstablishError::DialFailed(
                        format!("the leg-1 dial double is not part of the -R row: {target}"),
                    ))
                })
            });
        let double = fabric::wire_fabric_with_listen(fail_dial).await;

        let (work, key) = key_arg("tunnel-r");
        let mut serve = harness_serve_config(
            support::provider("tunnel-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        serve.channel_client = Some(Arc::clone(&double.client));
        serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
        let (ssh_port, accept_task) = tunnel_mount(serve).await;

        let fwd_port = support::pick_local_port().await;
        let mut ssh_child = ssh_forward_args(
            ssh_port,
            &key,
            &[
                "-R",
                &format!(
                    "127.0.0.1:{fwd_port}:{}:{}",
                    http_addr.ip(),
                    http_addr.port()
                ),
            ],
        )
        .spawn()
        .expect("vanilla ssh spawns");

        let through = tokio::time::timeout(Duration::from_secs(30), async {
            curl_until_success(&format!("http://127.0.0.1:{fwd_port}/"), &[]).await
        })
        .await;

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

        let out = through.expect("the -R row resolves in time");
        let stdout = String::from_utf8_lossy(&out.stdout).to_string();
        assert!(
            stdout.trim_end() == body.trim_end(),
            "vanilla ssh -R + curl: fabric arrival → forwarded-tcpip → client-dialed endpoint returns \
             the body — got {stdout:?}"
        );
        support::cleanup(&work);
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (`direct-tcpip`
    ///   serves **all** `ssh -D` traffic per POC-3 correction 2 — the
    ///   SOCKS handshake happens client-side in OpenSSH; POC-3 §B.2's
    ///   corrected posture re-landed with real curl).
    #[tokio::test]
    async fn real_ssh_dynamic_forward_serves_curl_through_the_direct_tcpip_leg() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("-D row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("curl", &["--version"]) {
            support::skip("-D row", "curl is unavailable");
            return;
        }

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

        let (work, key) = key_arg("tunnel-d");
        let mut serve = harness_serve_config(
            support::provider("tunnel-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        serve.channel_client = Some(fabric);
        let (ssh_port, accept_task) = tunnel_mount(serve).await;

        let socks_port = support::pick_local_port().await;
        let mut ssh_child =
            ssh_forward_args(ssh_port, &key, &["-D", &format!("127.0.0.1:{socks_port}")])
                .spawn()
                .expect("vanilla ssh spawns");

        let through = tokio::time::timeout(Duration::from_secs(30), async {
            let first = curl_until_success(
                &format!("http://{}:{}/", http_addr.ip(), http_addr.port()),
                &["--socks5-hostname", &format!("127.0.0.1:{socks_port}")],
            )
            .await;
            let second = curl_until_success(
                &format!("http://{}:{}/", http_addr.ip(), http_addr.port()),
                &["--socks5-hostname", &format!("127.0.0.1:{socks_port}")],
            )
            .await;
            (first, second)
        })
        .await;

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

        let (first, second) = through.expect("the -D row resolves in time");
        for out in [first, second] {
            let stdout = String::from_utf8_lossy(&out.stdout).to_string();
            assert!(
                stdout.trim_end() == body.trim_end(),
                "curl --socks5-hostname through vanilla ssh -D rides the door's direct-tcpip leg — \
                 got {stdout:?}"
            );
        }
        support::cleanup(&work);
    }
}

/// The decline-observability row where the `session` feature is off —
/// a real `ssh` pty attempt meets a clean, channel-scoped decline
/// (auth intact, no hang), with the door's decline event on record.
#[cfg(unix)]
#[cfg(not(feature = "session"))]
mod unfeatured_session_decline {
    use std::time::Duration;

    use alkssh::Limits;

    use super::common::event_capture::{expected_build, install};
    use super::common::fixture;
    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    /// - *Contract stamp:* mapping.md §Decline matrix (none/git/tunnel
    ///   builds decline session rows) + POC-1 §5 (decline
    ///   observability: channel-scoped decline, auth intact, no hangs).
    #[tokio::test]
    async fn real_openssh_pty_attempt_declines_cleanly_in_a_feature_limited_build() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("pty decline row", "ssh is unavailable");
            return;
        }

        install();
        let key = support::write_key("pty-decline");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let serve = harness_serve_config(
            support::provider("decline-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        let (port, accept_task) = support::mount(serve).await;

        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            tokio::process::Command::new("ssh")
                .args(["-tt", "-p", &port.to_string(), "-i"])
                .arg(&key)
                .args(support::ssh_options())
                .arg("decline-rider@127.0.0.1")
                .arg("echo unreachable")
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .output()
                .await
                .expect("ssh spawns")
        })
        .await;

        accept_task.abort();

        let output = match outcome {
            Ok(output) => output,
            Err(_) => panic!("the pty decline row hung past its 30 s deadline"),
        };
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();
        assert!(
            !output.status.success(),
            "the unfeatured pty attempt surfaces a failure to the real client — stderr: {stderr:?}"
        );
        assert!(
            !stderr.trim().is_empty(),
            "the decline surfaces a clean diagnostic — stderr: {stderr:?}"
        );
        if cfg!(feature = "git") {
            let event = support::decline_event(&[
                ("row", "pty_request"),
                ("message", "decline: channel request rejected"),
                ("feature_build", expected_build()),
            ])
            .await;
            assert_eq!(event.field("decline_class"), Some("request"));
            assert_eq!(event.field("reason"), Some("unfeatured"));
        } else {
            let event = support::decline_event(&[
                ("row", "session"),
                ("message", "decline: channel open rejected"),
                ("feature_build", expected_build()),
            ])
            .await;
            assert_eq!(event.field("decline_class"), Some("open"));
            assert_eq!(event.field("reason"), Some("unfeatured"));
        }
        support::cleanup(&work);
    }
}

/// The decline-observability rows where the `tunnel` feature is off —
/// a real `ssh -R` dies at the listen gate with a clean diagnostic;
/// a real `ssh -L`'s curls fail per connection without hanging the
/// client; the door's decline events hold for both.
#[cfg(unix)]
#[cfg(not(feature = "tunnel"))]
mod unfeatured_tunnel_declines {
    use std::time::Duration;

    use alkssh::Limits;

    use super::common::event_capture::{expected_build, install};
    use super::common::fixture;
    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    async fn plain_mount() -> (u16, tokio::task::JoinHandle<()>) {
        let serve = harness_serve_config(
            support::provider("decline-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        support::mount(serve).await
    }

    fn spawn_forward(forward: &[&str], port: u16, key: &std::path::Path) -> tokio::process::Child {
        tokio::process::Command::new("ssh")
            .arg("-N")
            .arg("-i")
            .arg(key)
            .arg("-p")
            .arg(port.to_string())
            .args(support::ssh_options())
            .args(forward)
            .arg("decline-rider@127.0.0.1")
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .expect("vanilla ssh spawns")
    }

    /// - *Contract stamp:* mapping.md §Global-request rows (the gate:
    ///   unfeatured `tcpip_forward` declines — POC-3 §C.2's refusal
    ///   shape re-landed with a real OpenSSH client and
    ///   `ExitOnForwardFailure`). The -R row is gated here
    ///   (check-gated, skipped-with-reason, not silently green).
    #[tokio::test]
    async fn real_openssh_remote_forward_attempt_declines_cleanly() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("-R decline row", "ssh is unavailable");
            return;
        }

        install();
        let key = support::write_key("rfwd-decline");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let (port, accept_task) = plain_mount().await;
        let listen_port = support::pick_local_port().await;

        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            let child = spawn_forward(
                &[
                    "-o",
                    "ExitOnForwardFailure=yes",
                    "-R",
                    &format!("127.0.0.1:{listen_port}:127.0.0.1:9365"),
                ],
                port,
                &key,
            );
            child.wait_with_output().await.expect("ssh completes")
        })
        .await;

        accept_task.abort();

        let output = match outcome {
            Ok(output) => output,
            Err(_) => panic!("the -R decline row hung past its 30 s deadline"),
        };
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();
        assert!(
            !output.status.success(),
            "the declined listen surfaces a failure to the real client — stderr: {stderr:?}"
        );
        assert!(
            stderr.contains("forwarding failed"),
            "OpenSSH's remote-forward-failed diagnostic is on record — stderr: {stderr:?}"
        );
        support::decline_event(&[
            ("row", "tcpip_forward"),
            ("message", "decline: global request rejected"),
            ("feature_build", expected_build()),
        ])
        .await;
        support::cleanup(&work);
    }

    /// - *Contract stamp:* mapping.md §Channel-type rows (the
    ///   `direct-tcpip` open gate — unfeatured declines; POC-1 §5:
    ///   the real client observes a clean per-connection failure,
    ///   no hang).
    #[tokio::test]
    async fn real_openssh_local_forward_connections_decline_per_connection() {
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("-L decline row", "ssh is unavailable");
            return;
        }
        if !support::tool_present("curl", &["--version"]) {
            support::skip("-L decline row", "curl is unavailable");
            return;
        }

        install();
        let key = support::write_key("lfwd-decline");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let (port, accept_task) = plain_mount().await;
        let fwd_port = support::pick_local_port().await;

        let mut ssh_child = spawn_forward(
            &["-L", &format!("127.0.0.1:{fwd_port}:127.0.0.1:9365")],
            port,
            &key,
        );
        let want: [(&str, &str); 3] = [
            ("row", "direct-tcpip"),
            ("message", "decline: channel open rejected"),
            ("feature_build", expected_build()),
        ];
        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            let deadline = std::time::Instant::now() + Duration::from_secs(28);
            loop {
                let out = tokio::process::Command::new("curl")
                    .args(["-s", "--max-time", "5"])
                    .arg(format!("http://127.0.0.1:{fwd_port}/"))
                    .output()
                    .await
                    .expect("curl spawns");
                assert!(
                    !out.status.success(),
                    "the declined direct-tcpip open must fail the forward, got exit {:?}",
                    out.status.code()
                );
                match super::common::event_capture::captured_with(&want) {
                    Some(event) => return event,
                    None => {
                        assert!(
                            std::time::Instant::now() < deadline,
                            "no direct-tcpip decline event within the row deadline"
                        );
                        tokio::time::sleep(Duration::from_millis(200)).await;
                    }
                }
            }
        })
        .await;

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

        let event = match outcome {
            Ok(event) => event,
            Err(_) => panic!("the -L decline row hung past its 30 s deadline"),
        };
        assert_eq!(event.field("decline_class"), Some("open"));
        assert_eq!(event.field("reason"), Some("unfeatured"));
        support::cleanup(&work);
    }
}

/// The decline-observability row for the sftp subsystem decline
/// (OQ-13) — a real `sftp` client's subsystem request meets the
/// named, channel-scoped decline with a clean diagnostic and no hang.
///
/// Row placement note: the sftp slot gates nothing, and the row can
/// only be exercised where session channels open (git / session
/// builds — in the none/tunnel builds the session-open decline is the
/// observable, already pinned above and by the matrix suite).
#[cfg(unix)]
#[cfg(any(feature = "session", feature = "git"))]
mod subsystem_decline {
    use std::time::Duration;

    use alkssh::Limits;

    use super::common::event_capture::{expected_build, install};
    use super::common::fixture;
    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    /// - *Contract stamp:* mapping.md §Channel-type rows (the
    ///   subsystem row: the `sftp` name declines-until-alkfs, OQ-13;
    ///   POC-3 §A.3's forced decline re-landed from a real sftp
    ///   client).
    #[tokio::test]
    async fn real_sftp_subsystem_attempt_declines_with_reason() {
        if !support::tool_present("sftp", &["-h"]) {
            support::skip("sftp decline row", "sftp is unavailable");
            return;
        }

        install();
        let key = support::write_key("sftp-decline");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let serve = harness_serve_config(
            support::provider("decline-rider"),
            &[fixture("bravo")],
            Limits::default(),
        );
        let (port, accept_task) = support::mount(serve).await;

        let batch = work.join("sftp-batch");
        std::fs::write(&batch, "ls\n").expect("the sftp batch file writes");

        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            tokio::process::Command::new("sftp")
                .arg("-b")
                .arg(&batch)
                .arg("-P")
                .arg(port.to_string())
                .arg("-i")
                .arg(&key)
                .args(support::ssh_options())
                .arg("decline-rider@127.0.0.1")
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::piped())
                .stderr(std::process::Stdio::piped())
                .output()
                .await
                .expect("sftp spawns")
        })
        .await;

        accept_task.abort();

        let output = match outcome {
            Ok(output) => output,
            Err(_) => panic!("the sftp decline row hung past its 30 s deadline"),
        };
        let stderr = String::from_utf8_lossy(&output.stderr).to_string();
        assert!(
            !output.status.success(),
            "the sftp decline surfaces a failure to the real client — stderr: {stderr:?}"
        );
        assert!(
            !stderr.trim().is_empty(),
            "the decline surfaces a clean diagnostic — stderr: {stderr:?}"
        );
        let event = support::decline_event(&[
            ("row", "subsystem_request"),
            ("message", "decline: channel request rejected"),
            ("feature_build", expected_build()),
        ])
        .await;
        assert_eq!(event.field("reason"), Some("unimplemented"));
        support::cleanup(&work);
    }
}

/// The decline-observability row for the unknown-repo collapsed
/// generic denial — a real `git clone` of a nonexistent repo dies
/// with the collapsed `access denied` band before any protocol byte
/// (no existence oracle) and exit 128.
#[cfg(unix)]
#[cfg(feature = "git")]
mod unknown_repo_decline {
    use std::sync::Arc;
    use std::time::Duration;

    use alkssh::Limits;

    use super::common::git_doubles as gd;
    use super::common::harness::serve_config as harness_serve_config;
    use super::support;

    /// - *Contract stamp:* mapping.md §Session-channel dispatch
    ///   (unknown repo → the collapsed generic denial *before any
    ///   protocol byte*; POC-3 §A.5's vanilla diagnostics: exit 128,
    ///   `fatal: remote error: access denied: owner/absent`).
    #[tokio::test]
    async fn real_git_clone_of_an_unknown_repo_collapses_to_the_generic_denial() {
        let (git_ok, git_reason) = support::git_available();
        if !git_ok {
            support::skip("unknown-repo row", &git_reason);
            return;
        }
        if !support::tool_present("ssh", &["-V"]) {
            support::skip("unknown-repo row", "ssh is unavailable");
            return;
        }

        let key = support::write_key("repo-denial");
        let work = key
            .parent()
            .expect("the workdir sits above the key")
            .to_path_buf();

        let doubles = Arc::new(gd::MemGit::new());
        let provider = support::provider("git-rider");
        let mut serve = harness_serve_config(
            provider,
            &[super::common::fixture("bravo")],
            Limits::default(),
        );
        serve.git_seam = Some(gd::git_seam(&doubles));
        let (port, accept_task) = support::mount(serve).await;

        let ssh = support::git_ssh_command(&key);
        let outcome = tokio::time::timeout(Duration::from_secs(30), async {
            tokio::process::Command::new("git")
                .arg("clone")
                .arg(format!("ssh://git-rider@127.0.0.1:{port}/owner/absent.git"))
                .arg(work.join("absent"))
                .env("GIT_SSH_COMMAND", &ssh)
                .output()
                .await
                .expect("git clone spawns")
        })
        .await;

        accept_task.abort();

        let outcome = match outcome {
            Ok(outcome) => outcome,
            Err(_) => panic!("the unknown-repo row hung past its 30 s deadline"),
        };
        let stderr = String::from_utf8_lossy(&outcome.stderr).to_string();
        assert_eq!(
            outcome.status.code(),
            Some(128),
            "the collapsed denial is git's fatal-remote-error exit — stderr: {stderr:?}"
        );
        assert!(
            stderr.contains("access denied") && stderr.contains("owner/absent"),
            "the denial names the collapsed `access denied` band with the normalized id, no existence \
             oracle — stderr: {stderr:?}"
        );
        assert!(
            !stderr.contains("not found"),
            "no repository-existence oracle leaks — stderr: {stderr:?}"
        );
        support::cleanup(&work);
    }
}