//! The in-process fake `sshd` (`w4-dial`), the **client-direction**
//! contract-test vehicle: a hand-rolled russh server handler standing
//! in for the vanilla `sshd` every wave-4 dial row is pointed at (the
//! reverse-runner shape). The alkssh loopback harness
//! ([`super::harness`]) drives the *server* direction and is not the
//! vehicle here (the client dials *vanilla* servers, so this double
//! is a plain russh server (no alkssh in its serving path), on a real
//! loopback TCP listener: a dial target is an address the dial
//! surface resolves and connects to, which a duplex pair has no
//! endpoint to rehearse.
//!
//! **Shape decisions (recorded, the task's placement item):**
//!
//! - **Where**: sibling of the wave-1 doubles under `tests/common/`;
//!   the module-placing convention (test surface, test-versioned
//!   freely, no public API export, no feature gate; the client
//!   handler skeleton it exercises compiles in every build).
//! - **One dial per instance**: doubles are cheap; build a fresh one
//!   per row. The listener accepts, serves the connection to
//!   completion on its own task, and loops (a row never re-dials the
//!   same double).
//! - **Identity**: the double resolves fingerprints against an
//!   injected accept set ([`SshdAccepts`]; exact-string equality,
//!   the family's only resolution shape, ADR-004 decision 1, the
//!   vanilla-parity stand-in for a real `sshd`'s
//!   `authorized_keys`). Every other auth hook is an explicit reject,
//!   never an inherited default (the ADR-004 decision 3 hygiene).
//! - **Host keys**: the fixture host-key pair rides the
//!   `russh::server::Config`; the presented fingerprint is the
//!   crate's single fingerprint path value (the rows' pin rules match on
//!   match on `fixture("bravo")`'s recorded value.
//! - **The session handle**: the double's `auth_succeeded` hook hands
//!   the per-connection [`russh::server::Handle`] to the row through
//!   a one-shot carrier ([`SshdDouble::session_handle`]), the row's
//!   initiator for the server-initiated-open decline rows (server →
//!   client `channel_open_*`; the dialed client's skeleton declines;
//!   the server's open call observes
//!   `Err(ChannelOpenFailure(..))`, the wire rows' observable).
//! - **The session handle**: the double's `auth_succeeded` hook hands
//!   the per-connection [`russh::server::Handle`] to the row through
//!   a one-shot carrier ([`SshdDouble::session_handle`]), the row's
//!   initiator for the server-initiated-open decline rows (server →
//!   client `channel_open_*`; the dialed client's skeleton declines;
//!   the server's open call observes
//!   `Err(ChannelOpenFailure(..))`, the wire rows' observable).
//! - **The acceptor stub**: rows that need a served session channel
//!   (the stays-usable-after-declines row here; the op-row vehicles
//!   later) arm the double to accept client session opens and hand
//!   the channel out ([`SshdDouble::next_opened_session`]).
//! - **The op-row script** (`w4-op-rows`): the double's
//!   request/open surfaces grow a small script: the session
//!   requests' reply (success/failure), the scripted stdout/stderr
//!   data + exit report a launched program would produce, and the
//!   direct opens' echo-or-reject behavior, plus an observation
//!   surface (the recorded commands/pty params/window
//!   changes/signals the client rows assert the wire verbatim
//!   against). The double stays vanilla-parity: a hand-rolled
//!   responder standing in for `sshd`, nothing alkssh-typed inside.
//!
//! - **The `-R` server half** (`w4-reverse-listen`): the double grows
//!   the reverse-forward surfaces a vanilla `sshd` really serves: the
//!   `tcpip-forward` / `streamlocal-forward` global-request hooks
//!   (scripted accept/deny; an accept **binds a real socket** in
//!   sshd's role, so a `port = 0` request's minted bound port is the
//!   socket's real one and the data leg is a genuine
//!   `ssh -R`-shaped round trip), the accept loop per bound listener
//!   (per accepted connection: a server-initiated `forwarded-*` open
//!   toward the client echoing the listen's own args + the
//!   connection's peer as the originator, then the bytes pumped
//!   bidirectionally; a client-declined open closes the accepted
//!   connection, vanilla parity), and the cancel forms removing the
//!   matching entry (tearing the real listener down; an unknown
//!   listener denies, sshd parity). The bookkeeping is per-connection
//!   handler state and bounded by the scripted listens alone (the
//!   double is a scripted stand-in, not a real server: at most one
//!   forward per scripted accept is what the rows drive.

#![allow(dead_code)]

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

use parking_lot::Mutex;
use russh::keys::PrivateKey;
use russh::server::{Auth, Handle, Msg, PublicKeyOffer, Response, Session, VerifiedPublicKey};
use russh::{Channel, ChannelOpenFailure, MethodKind, MethodSet};

/// The double's scripted exit report (what a launched program's
/// death would produce: `exit-status` / `exit-signal`).
#[derive(Debug, Clone, Copy)]
pub enum DblExit {
    /// `exit-status(N)`.
    Status(u32),
    /// `exit-signal(name)`, the wire's signal-name class.
    Signal(&'static str),
}

/// The double's scripted behavior for a client-initiated direct open
/// (`direct-tcpip` / `direct-streamlocal`).
#[derive(Debug, Clone, Copy)]
pub enum DblDirect {
    /// Reject the open: the scripted server-refusal class
    /// (`ConnectFailed`, the vanilla sshd's dead-target shape).
    Reject,
    /// Accept the open and echo-pump the channel (the pump rows'
    /// round-trip shape).
    Echo,
}

/// The double's scripted behavior for the client's reverse-forward
/// listen requests (`tcpip-forward` / `streamlocal-forward@…`), the
/// `-R` rows' scripted class (`w4-reverse-listen`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum DblListen {
    /// Deny the listen; the vanilla sshd's refusal (the scripted
    /// server-refusal class; the wire `RequestFailure` the client's
    /// op row mirrors). The script's default (fail-closed double).
    #[default]
    Deny,
    /// Accept the listen: the double binds a **real socket** in
    /// sshd's role (the actually-bound port rides the reply's mint,
    /// the port-0 row's real shape), records it for the cancel forms,
    /// and spawns the accept loop (per accepted connection: a
    /// server-initiated `forwarded-*` open toward the client echoing
    /// the listen's own args + the connection's peer, then the bytes
    /// pumped bidirectionally; a client-declined open closes that
    /// accepted connection, vanilla parity).
    Accept,
}

/// The fake `sshd`'s op-row script (the `w4-op-rows` request
/// surfaces), what the double answers and, on a success, what the
/// scripted "program" produces.
#[derive(Debug, Clone, Copy)]
pub struct SshdScript {
    /// The session requests' reply (`exec`/`pty`/`shell`):
    /// `true` → `channel_success`, `false` → `channel_failure` (the
    /// request-refusal rows' scripted class).
    pub request_success: bool,
    /// The channel data written as the program's stdout, after the
    /// request's success reply.
    pub stdout: &'static [u8],
    /// The channel data written as `extended_data(1)` stderr.
    pub stderr: &'static [u8],
    /// The exit report after the outputs (`None` = none, the
    /// no-report rows).
    pub exit: Option<DblExit>,
    /// The `direct-tcpip` opens' behavior.
    pub direct: DblDirect,
    /// The `direct-streamlocal` opens' behavior.
    pub streamlocal: DblDirect,
    /// The `tcpip-forward` listen form's behavior (`tcp -R` rows).
    pub tcp_listen: DblListen,
    /// The `streamlocal-forward` listen form's behavior (the unix
    /// `-R` rows).
    pub unix_listen: DblListen,
    /// The counted refusal; the double declines the *first N* client
    /// session opens (per connection, armed by
    /// `accept_session_channels`): the stays-usable-after-a-decline
    /// rows' scripted shape (one connection, first open refused, the
    /// next served).
    pub reject_first_session_opens: usize,
}

impl Default for SshdScript {
    /// The plain dial rows' posture: session opens accepted (the
    /// `accept_session_channels` flag governs), requests succeed with
    /// no outputs and no exit, direct opens rejected, listen forms
    /// denied (the scripted fail-closed default).
    fn default() -> Self {
        Self {
            request_success: true,
            stdout: &[],
            stderr: &[],
            exit: None,
            direct: DblDirect::Reject,
            streamlocal: DblDirect::Reject,
            tcp_listen: DblListen::Deny,
            unix_listen: DblListen::Deny,
            reject_first_session_opens: 0,
        }
    }
}

/// One recorded global-request listen/cancel form: (the request
/// name, the wire endpoint string, the wire port); the reverse-listen
/// rows assert the listens the client ops put on the wire.
pub type ListenRecord = (&'static str, String, u32);

/// The double's per-connection observations; the wire rows assert
/// the verbatim translations against these (what the client's op
/// rows put on the wire).
#[derive(Clone, Debug, Default)]
pub struct SshdObserved {
    execs: Arc<Mutex<Vec<ExecRecord>>>,
    ptys: Arc<Mutex<Vec<PtyRecord>>>,
    shells: Arc<Mutex<Vec<()>>>,
    windows: Arc<Mutex<Vec<WindowRecord>>>,
    signals: Arc<Mutex<Vec<String>>>,
    listens: Arc<Mutex<Vec<ListenRecord>>>,
    cancels: Arc<Mutex<Vec<ListenRecord>>>,
}

/// One recorded pty request's translated shapes: (term, cols, rows,
/// pixel_width, pixel_height).
pub type PtyRecord = (String, u32, u32, u32, u32);

/// One recorded window change's shapes: (cols, rows, pixel_width,
/// pixel_height).
pub type WindowRecord = (u32, u32, u32, u32);

/// One recorded exec request's command, verbatim bytes.
pub type ExecRecord = Vec<u8>;

impl SshdObserved {
    /// The exec commands seen (`exec_request`'s wire string,
    /// verbatim bytes).
    pub fn execs(&self) -> Vec<Vec<u8>> {
        self.execs.lock().clone()
    }

    /// The pty requests seen (term, cols, rows, pixel dims).
    pub fn ptys(&self) -> Vec<PtyRecord> {
        self.ptys.lock().clone()
    }

    /// Whether any `shell_request` was seen.
    pub fn shells(&self) -> Vec<()> {
        self.shells.lock().clone()
    }

    /// The window changes seen (cols, rows, pixel dims).
    pub fn windows(&self) -> Vec<WindowRecord> {
        self.windows.lock().clone()
    }

    /// The signals seen (russh's `Sig` debug names, "TERM",
    /// `Custom("..")`, …).
    pub fn signals(&self) -> Vec<String> {
        self.signals.lock().clone()
    }

    /// The global-request listen forms seen (name, endpoint, port;
    /// the requested port, never the mint).
    pub fn listens(&self) -> Vec<ListenRecord> {
        self.listens.lock().clone()
    }

    /// The global-request cancel forms seen (name, endpoint, port).
    pub fn cancels(&self) -> Vec<ListenRecord> {
        self.cancels.lock().clone()
    }
}

/// The double's scripted program serve, after a request's success
/// reply: stdout, then `extended_data(1)` stderr, then the exit
/// report, then the close (the vanilla sshd's exit-then-close
/// discipline).
async fn scripted_program(handle: Handle, channel_id: russh::ChannelId, script: Arc<SshdScript>) {
    if !script.stdout.is_empty()
        && handle
            .data(channel_id, script.stdout.to_vec())
            .await
            .is_err()
    {
        return;
    }
    if !script.stderr.is_empty()
        && handle
            .extended_data(channel_id, 1, script.stderr.to_vec())
            .await
            .is_err()
    {
        return;
    }
    match script.exit {
        Some(DblExit::Status(code)) => {
            let _ = handle.exit_status_request(channel_id, code).await;
        }
        Some(DblExit::Signal(name)) => {
            let sig = match name {
                "TERM" => russh::Sig::TERM,
                "INT" => russh::Sig::INT,
                other => russh::Sig::Custom(other.to_owned()),
            };
            let _ = handle
                .exit_signal_request(channel_id, sig, false, String::new(), "en".to_owned())
                .await;
        }
        None => {}
    }
    let _ = handle.close(channel_id).await;
}

/// The direct opens' echo pump, the accepted-typed-open shape's
/// server half, POC-1 §7's byte round-trip.
async fn direct_echo(mut stream: russh::ChannelStream<Msg>) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    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;
                }
            }
        }
    }
}

/// The `-R` accept loop per bound listener (the double's sshd role):
/// per accepted connection a server-initiated `forwarded-*` open
/// toward the client; the connected address echoes the listen
/// request's wire string verbatim, the port the actually-bound one,
/// and the originator the accepted connection's peer; then the bytes
/// pumped bidirectionally between the SSH channel and the accepted
/// socket. A client-declined open closes that accepted connection
/// (vanilla parity: the listener stays, the data has nowhere to go).
/// The loop ends with the listener itself (cancel tears it down; the
/// session's end fails the opens' sends).
async fn accept_forward_loop(
    handle: Handle,
    listener: tokio::net::TcpListener,
    connected_address: String,
    connected_port: u32,
) {
    loop {
        let Ok((conn, peer)) = listener.accept().await else {
            return;
        };
        let opened = handle
            .channel_open_forwarded_tcpip(
                connected_address.clone(),
                connected_port,
                peer.ip().to_string(),
                u32::from(peer.port()),
            )
            .await;
        match opened {
            Ok(channel) => {
                tokio::spawn(forward_connection(channel, conn));
            }
            Err(_) => drop(conn),
        }
    }
}

/// The accepted forward connection's pump (its SSH channel ↔ its
/// socket), the data leg's byte plane.
async fn forward_connection(channel: Channel<Msg>, mut conn: tokio::net::TcpStream) {
    let mut stream = channel.into_stream();
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await;
}

/// The unix accept loop's twin (the unix `-R` rows' shape): per
/// accepted socket a `forwarded-streamlocal` open toward the client
/// echoing the listen path.
#[cfg(unix)]
async fn accept_unix_forward_loop(
    handle: Handle,
    listener: tokio::net::UnixListener,
    socket_path: String,
) {
    loop {
        let Ok((conn, _)) = listener.accept().await else {
            return;
        };
        let opened = handle
            .channel_open_forwarded_streamlocal(socket_path.clone())
            .await;
        match opened {
            Ok(channel) => {
                tokio::spawn(unix_forward_connection(channel, conn));
            }
            Err(_) => drop(conn),
        }
    }
}

#[cfg(unix)]
async fn unix_forward_connection(channel: Channel<Msg>, mut conn: tokio::net::UnixStream) {
    let mut stream = channel.into_stream();
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await;
}

/// The fake `sshd`'s per-connection handler: accept-set identity
/// resolution, explicit rejects everywhere, the `auth_succeeded`
/// handle carrier and the session-open acceptor stub.
struct SshdDoubleHandler {
    accepts: Arc<parking_lot::RwLock<Vec<String>>>,
    accept_session_channels: bool,
    sessions_out: Arc<Mutex<Option<tokio::sync::oneshot::Sender<Handle>>>>,
    opened_sessions_out: tokio::sync::mpsc::UnboundedSender<Channel<Msg>>,
    script: Arc<SshdScript>,
    observed: SshdObserved,
    remaining_declines: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    /// The `-R` bookkeeping (per-connection, sshd's listener table):
    /// the key is the cancel-form's match key (the wire endpoint + the
    /// bound port (a `port = 0` mint's cancel matches the bound one),
    /// the value the live accept loop (cancel aborts it, dropping the
    /// real listener with it).
    tcp_forwards: std::collections::HashMap<(String, u32), tokio::task::JoinHandle<()>>,
    unix_forwards: std::collections::HashMap<String, tokio::task::JoinHandle<()>>,
}

impl russh::server::Handler for SshdDoubleHandler {
    type Error = russh::Error;

    /// The accept-set resolution: a verified key whose fingerprint is
    /// registered accepts (ADR-004 decision 1's shape, the vanilla
    /// `authorized_keys` stand-in); anything else rejects fail-closed.
    async fn auth_publickey(
        &mut self,
        _user: &str,
        key: VerifiedPublicKey<'_>,
    ) -> Result<Auth, Self::Error> {
        let fingerprint = alkssh::fingerprint(key.public_key);
        let accepted = self.accepts.read().contains(&fingerprint);
        Ok(if accepted {
            Auth::Accept
        } else {
            Auth::reject()
        })
    }

    /// Explicit accept-and-delegate (the probe hook consumes no auth
    /// attempt, POC-1 §4); the real disposition is
    /// [`Self::auth_publickey`]'s.
    async fn auth_publickey_offered(
        &mut self,
        _user: &str,
        _offer: PublicKeyOffer<'_>,
    ) -> Result<Auth, Self::Error> {
        Ok(Auth::Accept)
    }

    /// Explicit over inherited (ADR-004 decision 3 hygiene).
    async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
        Ok(Auth::reject())
    }

    /// Explicit over inherited.
    async fn auth_password(&mut self, _user: &str, _password: &str) -> Result<Auth, Self::Error> {
        Ok(Auth::reject())
    }

    /// Explicit over inherited.
    async fn auth_keyboard_interactive(
        &mut self,
        _user: &str,
        _submethods: &str,
        _response: Option<Response<'_>>,
    ) -> Result<Auth, Self::Error> {
        Ok(Auth::reject())
    }

    /// The handle carrier: the row initiates server-side opens
    /// through the returned [`Handle`].
    async fn auth_succeeded(&mut self, session: &mut Session) -> Result<(), Self::Error> {
        if let Some(carrier) = self.sessions_out.lock().take() {
            let _ = carrier.send(session.handle());
        }
        Ok(())
    }

    /// The acceptor stub: armed rows accept client session opens and
    /// hand the channel out; unarmed rows decline cleanly (the
    /// vehicle never accept-and-drops either); a counted refusal
    /// declines the first opens per connection (the
    /// stays-usable-after-a-decline rows' script).
    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        reply: russh::server::ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        let mut declined = false;
        while !declined {
            let current = self
                .remaining_declines
                .load(std::sync::atomic::Ordering::SeqCst);
            if current == 0 {
                break;
            }
            declined = self
                .remaining_declines
                .compare_exchange(
                    current,
                    current - 1,
                    std::sync::atomic::Ordering::SeqCst,
                    std::sync::atomic::Ordering::SeqCst,
                )
                .is_ok();
        }
        if declined {
            reply
                .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
                .await;
            return Ok(());
        }
        if self.accept_session_channels {
            self.opened_sessions_out
                .send(channel)
                .map_err(|_| russh::Error::SendError)?;
            reply.accept().await;
        } else {
            reply
                .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
                .await;
        }
        Ok(())
    }

    /// The `direct-tcpip` script: reject (the scripted
    /// server-refusal class `ConnectFailed`, the vanilla sshd's
    /// dead-target shape) or accept + echo pump.
    async fn channel_open_direct_tcpip(
        &mut self,
        channel: Channel<Msg>,
        _host_to_connect: &str,
        _port_to_connect: u32,
        _originator_address: &str,
        _originator_port: u32,
        reply: russh::server::ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        match self.script.direct {
            DblDirect::Reject => reply.reject(ChannelOpenFailure::ConnectFailed).await,
            DblDirect::Echo => {
                reply.accept().await;
                tokio::spawn(direct_echo(channel.into_stream()));
            }
        }
        Ok(())
    }

    /// The `direct-streamlocal` script (same behaviors).
    async fn channel_open_direct_streamlocal(
        &mut self,
        channel: Channel<Msg>,
        _socket_path: &str,
        reply: russh::server::ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        match self.script.streamlocal {
            DblDirect::Reject => reply.reject(ChannelOpenFailure::ConnectFailed).await,
            DblDirect::Echo => {
                reply.accept().await;
                tokio::spawn(direct_echo(channel.into_stream()));
            }
        }
        Ok(())
    }

    /// The pty script: the request is recorded and answered per the
    /// script (the request-refusal rows' class).
    #[allow(clippy::too_many_arguments)]
    async fn pty_request(
        &mut self,
        channel: russh::ChannelId,
        term: &str,
        col_width: u32,
        row_height: u32,
        pix_width: u32,
        pix_height: u32,
        _modes: &[(russh::Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.observed.ptys.lock().push((
            term.to_string(),
            col_width,
            row_height,
            pix_width,
            pix_height,
        ));
        self.reply_script_request(channel, session).await;
        Ok(())
    }

    /// The shell script: recorded + answered per the script; a
    /// success launches the scripted program.
    async fn shell_request(
        &mut self,
        channel: russh::ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.observed.shells.lock().push(());
        self.reply_script_request(channel, session).await;
        Ok(())
    }

    /// The exec script: the wire command is recorded verbatim (the
    /// client rows' construction/translation assertions), the reply
    /// per the script, and a success launches the scripted program.
    async fn exec_request(
        &mut self,
        channel: russh::ChannelId,
        command: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.observed.execs.lock().push(command.to_vec());
        self.reply_script_request(channel, session).await;
        Ok(())
    }

    /// The window-change observation (the pty control rows' translation
    /// witness).
    async fn window_change_request(
        &mut self,
        _channel: russh::ChannelId,
        col_width: u32,
        row_height: u32,
        pix_width: u32,
        pix_height: u32,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.observed
            .windows
            .lock()
            .push((col_width, row_height, pix_width, pix_height));
        Ok(())
    }

    /// The signal observation (names as russh carries them).
    async fn signal(
        &mut self,
        _channel: russh::ChannelId,
        signal: russh::Sig,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.observed.signals.lock().push(format!("{signal:?}"));
        Ok(())
    }

    /// The `tcpip-forward` listen form: recorded and scripted; an accept
    /// binds a real `127.0.0.1` listener in sshd's role (a `port = 0`
    /// request's mint is the socket's real bound port, the out-param
    /// parity), keys the listener by (wire address, bound port), and
    /// spawns the accept loop ([`accept_forward_loop`]).
    async fn tcpip_forward(
        &mut self,
        address: &str,
        port: &mut u32,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.observed
            .listens
            .lock()
            .push(("tcpip-forward", address.to_string(), *port));
        if self.script.tcp_listen == DblListen::Deny {
            return Ok(false);
        }
        let bind_target = format!("127.0.0.1:{}", *port);
        let Ok(listener) = tokio::net::TcpListener::bind(&bind_target).await else {
            return Ok(false);
        };
        let bound = listener
            .local_addr()
            .expect("the bound listener has an address")
            .port();
        let bound = u32::from(bound);
        if *port == 0 {
            *port = bound;
        }
        let task = tokio::spawn(accept_forward_loop(
            session.handle(),
            listener,
            address.to_string(),
            bound,
        ));
        self.tcp_forwards.insert((address.to_string(), bound), task);
        Ok(true)
    }

    /// The `cancel-tcpip-forward` form: matches by the raw wire
    /// strings (the bound port for a mint); a hit removes the
    /// listener (tearing the real socket down, the cancel row's
    /// observable), an unknown listener denies (sshd parity).
    async fn cancel_tcpip_forward(
        &mut self,
        address: &str,
        port: u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.observed
            .cancels
            .lock()
            .push(("cancel-tcpip-forward", address.to_string(), port));
        match self.tcp_forwards.remove(&(address.to_string(), port)) {
            Some(listener_task) => {
                listener_task.abort();
                Ok(true)
            }
            None => Ok(false),
        }
    }

    /// The `streamlocal-forward` listen form (the unix `-R`): records,
    /// scripts, binds the real unix socket (stale file unlinked
    /// first), and spawns the unix accept loop. Off-Unix the hook is
    /// not implemented and falls to the trait's deny default (the
    /// vanilla Windows sshd parity); the crate's own runtime-Unix
    /// gate mirrors the refusal client-side before the wire.
    #[cfg(unix)]
    async fn streamlocal_forward(
        &mut self,
        socket_path: &str,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.observed
            .listens
            .lock()
            .push(("streamlocal-forward", socket_path.to_string(), 0));
        if self.script.unix_listen == DblListen::Deny {
            return Ok(false);
        }
        let _ = std::fs::remove_file(socket_path);
        let Ok(listener) = tokio::net::UnixListener::bind(socket_path) else {
            return Ok(false);
        };
        let task = tokio::spawn(accept_unix_forward_loop(
            session.handle(),
            listener,
            socket_path.to_string(),
        ));
        self.unix_forwards.insert(socket_path.to_string(), task);
        Ok(true)
    }

    /// The `cancel-streamlocal-forward` form (matches by the path
    /// alone, the unix cancel's sshd parity).
    #[cfg(unix)]
    async fn cancel_streamlocal_forward(
        &mut self,
        socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        self.observed.cancels.lock().push((
            "cancel-streamlocal-forward",
            socket_path.to_string(),
            0,
        ));
        match self.unix_forwards.remove(socket_path) {
            Some(listener_task) => {
                listener_task.abort();
                Ok(true)
            }
            None => Ok(false),
        }
    }
}

impl SshdDoubleHandler {
    /// The scripted request reply: success launches the scripted
    /// program through the session's handle (spawned, so the packet
    /// loop returns fast) *when a program is scripted at all*; an
    /// output-less/no-exit script launches nothing (the interactive
    /// sessions stay open; close is the client's or the connection's);
    /// the failure class answers `channel_failure`.
    async fn reply_script_request(&self, channel: russh::ChannelId, session: &mut Session) {
        if !self.script.request_success {
            let _ = session.channel_failure(channel);
            return;
        }
        let _ = session.channel_success(channel);
        let has_program = !self.script.stdout.is_empty()
            || !self.script.stderr.is_empty()
            || self.script.exit.is_some();
        if has_program {
            let (handle, channel_id) = (session.handle(), channel);
            let script = Arc::clone(&self.script);
            tokio::spawn(scripted_program(handle, channel_id, script));
        }
    }
}

/// The fake `sshd`'s accepted-identity register, exact-string
/// fingerprints (the row registers the fixture dials with).
pub struct SshdAccepts {
    inner: Arc<parking_lot::RwLock<Vec<String>>>,
}

impl SshdAccepts {
    /// Registers one accepted fingerprint (the vanilla
    /// `authorized_keys` row's shape).
    pub fn register(&self, fingerprint: &str) {
        self.inner.write().push(fingerprint.to_string());
    }
}

/// One spawned fake `sshd`, the client-direction contract vehicle.
pub struct SshdDouble {
    /// The dial target, `<ip>:<port>`, the row's
    /// `DialConfig.address` value.
    pub address: String,
    session_carrier: Option<tokio::sync::oneshot::Receiver<Handle>>,
    opened_sessions: tokio::sync::mpsc::UnboundedReceiver<Channel<Msg>>,
    /// The per-connection serving outcomes (the clean-close rows'
    /// observable); `None` after the double's end.
    out_rx: Option<tokio::sync::mpsc::UnboundedReceiver<Result<(), russh::Error>>>,
    accept_task: tokio::task::JoinHandle<()>,
}

/// Spawns the double on `127.0.0.1:0` under the given op-row script:
/// `host_keys` rides the `russh::server::Config`;
/// `accept_session_channels` arms the session-open acceptor stub.
/// Returns the double, its accept-set register (register before
/// dialing, since the wire's auth reads it at the dial), and the
/// observation surface (see [`SshdObserved`]).
pub async fn spawn_scripted(
    host_keys: Vec<PrivateKey>,
    accept_session_channels: bool,
    script: SshdScript,
) -> (SshdDouble, SshdAccepts, SshdObserved) {
    let accepts = Arc::new(parking_lot::RwLock::new(Vec::new()));
    let (session_tx, session_rx) = tokio::sync::oneshot::channel();
    let (opened_tx, opened_rx) = tokio::sync::mpsc::unbounded_channel();
    let (outcomes_tx, outcomes_rx) = tokio::sync::mpsc::unbounded_channel();
    let sessions_out = Arc::new(Mutex::new(Some(session_tx)));
    let script = Arc::new(script);

    let config = Arc::new(russh::server::Config {
        methods: MethodSet::from(&[MethodKind::PublicKey][..]),
        keys: host_keys,
        inactivity_timeout: Some(Duration::from_secs(30)),
        ..Default::default()
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
        .await
        .expect("the double's loopback listener binds");
    let addr = listener.local_addr().expect("the double has an address");

    let accepts_for_loop = Arc::clone(&accepts);
    let observed = SshdObserved::default();
    let observed_for_loop = observed.clone();
    let script_declines = script.reject_first_session_opens;
    let accept_task = tokio::spawn(async move {
        loop {
            let (socket, _) = match listener.accept().await {
                Ok(accepted) => accepted,
                Err(_) => break,
            };
            let config = Arc::clone(&config);
            let handler = SshdDoubleHandler {
                accepts: Arc::clone(&accepts_for_loop),
                accept_session_channels,
                sessions_out: Arc::clone(&sessions_out),
                opened_sessions_out: opened_tx.clone(),
                script: Arc::clone(&script),
                observed: observed_for_loop.clone(),
                remaining_declines: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
                    script_declines,
                )),
                tcp_forwards: std::collections::HashMap::new(),
                unix_forwards: std::collections::HashMap::new(),
            };
            let outcomes = outcomes_tx.clone();
            tokio::spawn(async move {
                let session = match russh::server::run_stream(config, socket, handler).await {
                    Ok(session) => session,
                    Err(_) => {
                        let _ = outcomes.send(Err(russh::Error::Disconnect));
                        return;
                    }
                };
                let _ = outcomes.send(session.await);
            });
        }
    });

    (
        SshdDouble {
            address: format!("{}:{}", addr.ip(), addr.port()),
            session_carrier: Some(session_rx),
            opened_sessions: opened_rx,
            out_rx: Some(outcomes_rx),
            accept_task,
        },
        SshdAccepts { inner: accepts },
        observed,
    )
}

/// Spawns the double on `127.0.0.1:0` with the plain dial rows'
/// default script, the `w4-dial` shape kept: the double and its
/// accept-set register.
pub async fn spawn(
    host_keys: Vec<PrivateKey>,
    accept_session_channels: bool,
) -> (SshdDouble, SshdAccepts) {
    let (double, accepts, _observed) =
        spawn_scripted(host_keys, accept_session_channels, SshdScript::default()).await;
    (double, accepts)
}

impl SshdDouble {
    /// The dialed session's server-side handle, the initiator for
    /// the server-initiated-open rows. Awaits the `auth_succeeded`
    /// carrier; the dial has authenticated by the time a row calls
    /// this (the carrier's sender only fires inside the double's
    /// successful auth flow), so the await is prompt.
    pub async fn session_handle(&mut self) -> Handle {
        let carrier = self
            .session_carrier
            .take()
            .expect("the double's session handle arrives once (one dial per double)");
        carrier
            .await
            .expect("the double's session handle arrives (the row dialed and authenticated)")
    }

    /// The client-initiated session channels the acceptor stub
    /// accepted; `None` after the session's end (the observation
    /// shape's honest empty).
    pub async fn next_opened_session(&mut self) -> Option<Channel<Msg>> {
        self.opened_sessions.recv().await
    }

    /// The next per-connection serving outcome, the clean-close
    /// rows' observable (a clean `Disconnect::ByApplication` is the
    /// sessions' `Ok`; an abrupt peer close is an `Err` wire class).
    /// `None` after the double's loop ends.
    pub async fn next_session_outcome(&mut self) -> Option<Result<(), russh::Error>> {
        let receiver = self.out_rx.as_mut()?;
        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
        tokio::time::timeout_at(deadline, receiver.recv())
            .await
            .ok()
            .flatten()
    }

    /// Shuts the double down (the accept loop aborts; the connection,
    /// if still open, drops with it).
    pub fn shutdown(mut self) {
        self.out_rx.take();
        self.accept_task.abort();
    }
}

impl std::fmt::Debug for SshdDouble {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SshdDouble")
            .field("address", &self.address)
            .finish()
    }
}