alkssh/tests/common/harness.rs

//! alkssh's loopback harness (`w2-loopback-harness`) — the in-process
//! client/server pair that is *the* contract-test vehicle for the
//! crate: the wave-2 decline matrix, every wave-3 mapping's contract
//! tests, and the client-direction smokes all drive the crate through
//! this module.
//!
//! **Transport decision (recorded, the `w2-loopback-harness` flag):**
//! the server-side [`SshAdapter::handle`](alkssh::SshAdapter::handle)
//! mount runs over a **tokio duplex pair** ([`tokio::io::duplex`],
//! [`DUPLEX_CAPACITY`] bytes each direction), and the client side
//! connects with [`russh::client::connect_stream`] — russh's real
//! codec, kex and transport run over the pair verbatim, so the wire
//! path is fully exercised. The one thing it deliberately does not
//! carry is OS-socket semantics; TCP loopback (`127.0.0.1:0`) was the
//! alternative and is rejected: real-socket nondeterminism (listener
//! lifecycle, loopback timing, accept-loop flakiness under slow CI)
//! buys only what the real-tool smokes of the mapping waves and the
//! e2e wave already cover cheaply. The duplex pair is deterministic,
//! needs no port, and cannot collide — the no-TCP-port-dependency
//! shape the acceptance criteria name.
//!
//! **Placement decision (recorded, the `w2-loopback-harness`
//! counterpart of the wave-1 doubles' location decision):** the
//! harness lives as a sibling module of the wave-1 doubles, under
//! `tests/common/`, reached with the same `mod common;` convention —
//! test surface, test-versioned freely, never a public API export
//! (no re-export from `lib.rs`, no feature gate). Unlike the doubles
//! it is a *vehicle* convention, not just a seam: later waves'
//! contract tests do not re-derive a mount/connect pair; they call
//! [`drive`] (or compose its parts), so a transport revisit stays a
//! one-module change.
//!
//! **Client-side posture (recorded):** the wave-4 dial surface does
//! not exist yet; the harness drives a russh client directly with the
//! doubles: a fixture key ([`authenticate_publickey`]), the
//! [`MemHostKeyPolicy`] decision double bridged into russh's
//! `check_server_key` hook ([`HarnessClientHandler`]), and the
//! keepalive / wall-clock posture from [`alkssh::Limits`]
//! ([`client_config`]). The permissive posture for routine rows is
//! the double's `with_default(true)` — an explicit test choice; the
//! accept-all posture exists only here, never as a crate built-in,
//! and the fail-closed rows construct `with_default(false)`.
//!
//! **Host identity on the loopback (recorded):** a duplex pair has no
//! endpoint, so the presented host key is always labeled with
//! [`LOOPBACK_HOST`] and [`LOOPBACK_PORT`]; host-key policy rows pin
//! rules against those values verbatim. The presented key's
//! fingerprint is the crate's single fingerprint path
//! ([`alkssh::fingerprint`]) — exactly the decision input the wave-4
//! dial wiring will compute at russh's own hook, so this bridge is
//! that wiring's rehearsal.
//!
//! Helpers stay thin — the russh client API is the vocabulary; these
//! remove boilerplate (config assembly, mount/spawn, fixture-key
//! plumbing, drain loops), never semantics: every observation is a
//! russh-verbatim shape ([`AuthResult`], [`Error`],
//! [`ChannelMsg`]) the rows match on directly.
//!
//! The helper set grows with the rows it serves: `w2-loopback-harness`
//! landed the auth/session/exec/shell/subsystem vocabulary, and
//! `w2-dispatch-decline-core` extended it with the decline-matrix
//! vocabulary (the tunnel/x11 opens, the pty/env/x11/agent-forward
//! requests, the listen requests) — its recorded expectation that the
//! "session-request and exit-status/close helpers are exercised by
//! construction this wave ... their live rows arrive with
//! `w2-dispatch-decline-core`" is discharged by exactly that
//! extension, not by a harness rewrite.
//!
//! Like the doubles: test functions are *rows* carrying the greppable
//! **`Contract stamp:`** doc markers (see the common module's doc).

#![allow(dead_code)]

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

use alkcall::core::auth::IdentityProvider;
use parking_lot::Mutex;
use russh::client::{AuthResult, Handle, Msg};
use russh::keys::PrivateKeyWithHashAlg;
use russh::{Channel, ChannelId, ChannelMsg, ChannelOpenFailure, Disconnect, Error, MethodSet};

use alkssh::{
    AdapterError, HostKeyPolicy, Limits, PresentedHostKey, ServeConfig, SshAdapter, fingerprint,
};

use super::{FixtureKey, MemHostKeyPolicy, MemHostKeySource};

/// The host label the loopback presents to the host-key policy — a
/// duplex pair has no endpoint, so every decision is labeled with
/// this constant (see the module docs §Host identity).
pub const LOOPBACK_HOST: &str = "loopback";

/// The port label for the loopback's presented host key — `0`, the
/// no-port marker: the transport has no TCP port by design.
pub const LOOPBACK_PORT: u16 = 0;

/// Each duplex direction's buffer — the `w2-server-mount` wire-smoke
/// precedent's capacity, kept as the harness convention.
pub const DUPLEX_CAPACITY: usize = 65_536;

/// The harness's russh client config — the keepalive / wall-clock
/// posture from [`Limits`]: `keepalive_interval` / `keepalive_max`
/// mapped verbatim, `inactivity_timeout` carried as always-set (the
/// crate never constructs a session with no wall clock,
/// [`alkssh::Limits`]' in-crate posture extended to the harness's
/// client direction). Everything else stays russh's own default.
pub fn client_config(limits: Limits) -> Arc<russh::client::Config> {
    Arc::new(russh::client::Config {
        inactivity_timeout: Some(limits.inactivity_timeout),
        keepalive_interval: limits.keepalive_interval,
        keepalive_max: limits.keepalive_max,
        ..Default::default()
    })
}

/// One server-initiated forwarded open the rows observe at the
/// harness client (`w3-tunnel-listen`'s accept-hook harness row — the
/// `-R` data leg arrives here). The `Channel` only carries data after
/// the row accepts (the harness hook answers the open itself).
#[derive(Debug)]
pub enum ForwardedOpen {
    /// The `forwarded-tcpip` row — the echo args + originator the door
    /// opened with, verbatim as the wire carried them.
    Tcpip {
        connected_address: String,
        connected_port: u32,
        originator_address: String,
        originator_port: u32,
        channel: Channel<Msg>,
    },
    /// The `forwarded-streamlocal@openssh.com` row (the unix listen
    /// leg's arrival shape).
    Streamlocal {
        socket_path: String,
        channel: Channel<Msg>,
    },
}

/// The harness client's acceptor — what a vanilla client's matching
/// `-R` listener stands-in as: when installed, server-initiated
/// forwarded opens are accepted and handed to the row through the
/// channel the acceptor was built with; **no acceptor installed → the
/// open declines cleanly** (`AdministrativelyProhibited` — the
/// family's client-side contract, ADR-003 symmetry: a vanillla stand-
/// in with no matching forward never accept-and-drops).
pub struct ForwardedAcceptor {
    opens: tokio::sync::mpsc::UnboundedSender<ForwardedOpen>,
}

impl ForwardedAcceptor {
    /// Builds the acceptor over its event channel; the receiving half
    /// rides the rows (`drain_forwarded`).
    pub fn new() -> (Self, tokio::sync::mpsc::UnboundedReceiver<ForwardedOpen>) {
        let (opens, receiver) = tokio::sync::mpsc::unbounded_channel();
        (Self { opens }, receiver)
    }

    fn hand_off(&self, open: ForwardedOpen) -> Result<(), Error> {
        self.opens
            .send(open)
            .map_err(|_| Error::ChannelOpenFailure(ChannelOpenFailure::UnknownChannelType))
    }
}

/// The client-direction data gate (`w4-git-exit-close`'s row
/// vehicle): armed on a channel id, the harness client's packet loop
/// pends inside its `data` hook for that channel's packets (the hook
/// is awaited inline in russh's client loop) — the server's outbound
/// writes for the channel back up through the transport's duplex pair
/// into the server's message queue until [`DataGate::release`]. The
/// held counter (`DataGate::held_packets`) lets the row observe that
/// the freeze is real before it releases.
pub struct DataGate {
    target: Mutex<Option<ChannelId>>,
    opened: Mutex<Option<tokio::sync::oneshot::Sender<()>>>,
    release: Mutex<Option<tokio::sync::oneshot::Receiver<()>>>,
    held: AtomicUsize,
}

impl DataGate {
    /// Builds the inert gate — arm it with [`DataGate::arm`] once the
    /// row has the channel id to gate.
    pub fn new() -> Self {
        Self {
            target: Mutex::new(None),
            opened: Mutex::new(None),
            release: Mutex::new(None),
            held: AtomicUsize::new(0),
        }
    }

    /// Arms the gate on one channel id: the next client-direction
    /// data packet for that channel pends the client's packet loop.
    pub fn arm(&self, channel: ChannelId) {
        let (opened, release) = tokio::sync::oneshot::channel();
        *self.target.lock() = Some(channel);
        *self.opened.lock() = Some(opened);
        *self.release.lock() = Some(release);
    }

    /// How many data packets have hit the armed gate so far — the
    /// row's freeze signal (count `>= 1` means the client's packet
    /// loop is pended inside the hook).
    pub fn held_packets(&self) -> usize {
        self.held.load(Ordering::Acquire)
    }

    /// Releases the gate — the client's packet loop resumes and the
    /// backed-up server queues drain.
    pub fn release(&self) {
        *self.target.lock() = None;
        if let Some(opened) = self.opened.lock().take() {
            let _ = opened.send(());
        }
    }

    async fn block(&self, channel: ChannelId) {
        if *self.target.lock() != Some(channel) {
            return;
        }
        self.held.fetch_add(1, Ordering::Release);
        let release = self.release.lock().take();
        if let Some(release) = release {
            let _ = release.await;
        }
    }
}

/// The harness's russh client handler — the [`MemHostKeyPolicy`]
/// decision double bridged into russh's `check_server_key` hook, with
/// the decision input exactly as the wave-4 dial wiring will shape it
/// ([`PresentedHostKey`] over the crate's fingerprint path), plus the
/// `w3-tunnel-listen` accept hook: the server-initiated forwarded
/// opens the contract rows observe (the ADR-003 §4 client-side
/// acceptor symmetry).
pub struct HarnessClientHandler {
    policy: Arc<MemHostKeyPolicy>,
    acceptor: Option<Arc<ForwardedAcceptor>>,
    data_gate: Option<Arc<DataGate>>,
}

impl HarnessClientHandler {
    /// Builds the handler over a policy double — `with_default(true)`
    /// for the routine permissive posture, `with_default(false)` (or a
    /// pinned rule) for the fail-closed rows — with the forwarded
    /// acceptor absent (opens decline cleanly; install one with
    /// [`HarnessClientHandler::with_forwarded_acceptor`]).
    pub fn new(policy: Arc<MemHostKeyPolicy>) -> Self {
        Self {
            policy,
            acceptor: None,
            data_gate: None,
        }
    }

    /// Installs the acceptor (the matching-`-R`-listener stand-in).
    pub fn with_forwarded_acceptor(self, acceptor: Arc<ForwardedAcceptor>) -> Self {
        Self {
            policy: self.policy,
            acceptor: Some(acceptor),
            data_gate: self.data_gate,
        }
    }

    /// Installs the client-direction data gate (`w4-git-exit-close`'s
    /// row vehicle). Inert until the gate is armed on a channel id.
    pub fn with_data_gate(self, gate: Arc<DataGate>) -> Self {
        Self {
            policy: self.policy,
            acceptor: self.acceptor,
            data_gate: Some(gate),
        }
    }
}

impl russh::client::Handler for HarnessClientHandler {
    type Error = Error;

    /// The data gate's hook: when the gate is armed on this channel id
    /// and not yet released, the client's packet loop pends inside it
    /// (the hook is awaited inline) — the server's outbound writes
    /// back up through the transport into its message queue until the
    /// row releases. Inert otherwise; also the block counter the row
    /// polls to know the freeze is real.
    async fn data(
        &mut self,
        channel: ChannelId,
        _data: &[u8],
        _session: &mut russh::client::Session,
    ) -> Result<(), Self::Error> {
        if let Some(gate) = self.data_gate.as_ref() {
            gate.block(channel).await;
        }
        Ok(())
    }

    /// The policy decision, verbatim: the presented key's fingerprint
    /// is computed on the crate's single fingerprint path, the
    /// endpoint labels are the loopback constants, and the policy's
    /// `bool` verdict rides russh's hook straight through.
    async fn check_server_key(
        &mut self,
        server_public_key: &russh::keys::PublicKeyOrCertificate,
    ) -> Result<bool, Self::Error> {
        let presented = PresentedHostKey {
            host: LOOPBACK_HOST.into(),
            port: LOOPBACK_PORT,
            fingerprint: fingerprint(&server_public_key.public_key()),
        };
        Ok(self.policy.decide(&presented).await)
    }

    /// The `forwarded-tcpip` accept hook: accept-into-the-acceptor
    /// (the row observes the echoed args + originator verbatim and
    /// pumps the `Channel`), or the clean decline when no acceptor is
    /// installed.
    async fn server_channel_open_forwarded_tcpip(
        &mut self,
        channel: Channel<Msg>,
        connected_address: &str,
        connected_port: u32,
        originator_address: &str,
        originator_port: u32,
        reply: russh::client::ChannelOpenHandle,
        _session: &mut russh::client::Session,
    ) -> Result<(), Self::Error> {
        match self.acceptor.as_ref() {
            Some(acceptor) => {
                let open = ForwardedOpen::Tcpip {
                    connected_address: connected_address.to_owned(),
                    connected_port,
                    originator_address: originator_address.to_owned(),
                    originator_port,
                    channel,
                };
                acceptor.hand_off(open)?;
                reply.accept().await;
            }
            None => {
                reply
                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
                    .await;
            }
        }
        Ok(())
    }

    /// The `forwarded-streamlocal` accept hook: the unix leg's shape.
    async fn server_channel_open_forwarded_streamlocal(
        &mut self,
        channel: Channel<Msg>,
        socket_path: &str,
        reply: russh::client::ChannelOpenHandle,
        _session: &mut russh::client::Session,
    ) -> Result<(), Self::Error> {
        match self.acceptor.as_ref() {
            Some(acceptor) => {
                let open = ForwardedOpen::Streamlocal {
                    socket_path: socket_path.to_owned(),
                    channel,
                };
                acceptor.hand_off(open)?;
                reply.accept().await;
            }
            None => {
                reply
                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
                    .await;
            }
        }
        Ok(())
    }
}

/// Assembles the server side from the wave-1 doubles — a
/// [`ServeConfig`] over the given identity provider and fixture host
/// keys, with the session limits applied.
pub fn serve_config(
    provider: Arc<dyn IdentityProvider>,
    host_keys: &[&'static FixtureKey],
    limits: Limits,
) -> ServeConfig {
    let mut serve = ServeConfig::new(
        provider,
        Arc::new(MemHostKeySource::from_fixtures(host_keys)),
    );
    serve.limits = limits;
    serve
}

/// Constructs the adapter from the doubles — [`serve_config`] with the
/// default [`Limits`]; the decrypt-at-construction failure paths are
/// the construction tests' business (`w2-server-mount`), so a harness
/// row's construction failing is a bug, not an observation.
pub async fn adapter(
    provider: Arc<dyn IdentityProvider>,
    host_keys: &[&'static FixtureKey],
) -> SshAdapter {
    SshAdapter::new(serve_config(provider, host_keys, Limits::default()))
        .await
        .expect("adapter constructs (fixture host keys load)")
}

/// One driven loopback session: the mounted server task plus the
/// connected client handle and its policy double.
pub struct LoopbackSession {
    /// The connected russh client — the helpers' driver.
    pub client: Handle<HarnessClientHandler>,
    /// The policy double the connection's host-key decision rode —
    /// decision-log inspection for the rows.
    pub host_key_policy: Arc<MemHostKeyPolicy>,
    server_task: tokio::task::JoinHandle<Result<(), AdapterError>>,
}

impl LoopbackSession {
    /// Ends one loopback row cleanly: disconnects the client, waits
    /// for the server task, and requires a clean session end on both
    /// sides — the harness's every-row termination posture.
    pub async fn unmount(self) {
        self.client
            .disconnect(Disconnect::ByApplication, "row done", "en")
            .await
            .expect("client disconnects cleanly");
        let outcome = self.server_task.await.expect("server task ends");
        if let Err(err) = outcome {
            panic!("session end must be clean, got {err:?}");
        }
    }
}

/// Drives one loopback session: mounts the adapter over a duplex pair
/// (spawned to completion, the assembly's accept-loop shape), connects
/// a harness client against it and returns the pair. The adapter is
/// consumed (the spawned task owns it) — build a fresh one per row;
/// the doubles make that cheap.
///
/// Fallible at the connect step on purpose: a fail-closed host-key
/// policy verdict is a disposition the rows observe
/// (`Error::UnknownKey`); a routine row's `Err` is its bug.
pub async fn drive(
    adapter: SshAdapter,
    host_key_policy: Arc<MemHostKeyPolicy>,
    limits: Limits,
) -> Result<LoopbackSession, Error> {
    drive_with_acceptor(adapter, host_key_policy, limits, None).await
}

/// [`drive`] with the client-side forwarded acceptor installed — the
/// `w3-tunnel-listen` rows' vehicle (the harness handler's accept
/// hook consumes the installed acceptor; `None` = the clean-decline
/// posture [`drive`] carries).
pub async fn drive_with_acceptor(
    adapter: SshAdapter,
    host_key_policy: Arc<MemHostKeyPolicy>,
    limits: Limits,
    acceptor: Option<Arc<ForwardedAcceptor>>,
) -> Result<LoopbackSession, Error> {
    mount_and_connect(adapter, host_key_policy, limits, acceptor, None).await
}

/// [`drive`] with the client-direction data gate installed — the
/// `w4-git-exit-close` row's vehicle.
pub async fn drive_with_data_gate(
    adapter: SshAdapter,
    host_key_policy: Arc<MemHostKeyPolicy>,
    limits: Limits,
    gate: Option<Arc<DataGate>>,
) -> Result<LoopbackSession, Error> {
    mount_and_connect(adapter, host_key_policy, limits, None, gate).await
}

/// `[drive*]`'s shared core: mount + connect (the doc decisions at
/// [`drive`] cover both; the optional hooks ride the handler).
async fn mount_and_connect(
    adapter: SshAdapter,
    host_key_policy: Arc<MemHostKeyPolicy>,
    limits: Limits,
    acceptor: Option<Arc<ForwardedAcceptor>>,
    data_gate: Option<Arc<DataGate>>,
) -> Result<LoopbackSession, Error> {
    let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY);
    let server_task = tokio::spawn(async move { adapter.handle(server_io).await });

    let mut handler = HarnessClientHandler::new(Arc::clone(&host_key_policy));
    if let Some(acceptor) = acceptor {
        handler = handler.with_forwarded_acceptor(acceptor);
    }
    if let Some(gate) = data_gate {
        handler = handler.with_data_gate(gate);
    }
    let client = russh::client::connect_stream(client_config(limits), client_io, handler).await?;

    Ok(LoopbackSession {
        client,
        host_key_policy,
        server_task,
    })
}

/// Authenticates one fixture key — the auth helper: the fixture
/// plumbing (`Arc` + `PrivateKeyWithHashAlg`) removed; the
/// [`AuthResult`] comes back verbatim for the success / failure-shape
/// rows to match on.
pub async fn authenticate_publickey(
    client: &mut Handle<HarnessClientHandler>,
    username: &str,
    key: &FixtureKey,
) -> AuthResult {
    let private = Arc::new(key.private_key());
    client
        .authenticate_publickey(username, PrivateKeyWithHashAlg::new(private, None))
        .await
        .expect("the auth exchange completes over the wire")
}

/// The failure shape's advertisement half: the remaining-methods set
/// carried by an [`AuthResult::Failure`] — `None` when the result is
/// a success (the rows' success/failure observation shape).
pub fn remaining_methods(result: &AuthResult) -> Option<MethodSet> {
    match result {
        AuthResult::Failure {
            remaining_methods, ..
        } => Some(remaining_methods.clone()),
        _ => None,
    }
}

/// Opens a session channel — the open helper. Success is a confirmed
/// [`Channel`]; the open-failure shape is the `Err` verbatim
/// ([`Error::ChannelOpenFailure`] carries the SSH decline reason the
/// decline-matrix rows match on).
pub async fn open_session_channel(
    client: &Handle<HarnessClientHandler>,
) -> Result<Channel<Msg>, Error> {
    client.channel_open_session().await
}

/// Opens a `direct-tcpip` channel — the tunnel open helper (`ssh -L`
/// / `ssh -D` traffic). The open-failure shape is the `Err` verbatim,
/// the decline-matrix rows' reason surface.
pub async fn open_direct_tcpip(
    client: &Handle<HarnessClientHandler>,
    host_to_connect: &str,
    port_to_connect: u32,
) -> Result<Channel<Msg>, Error> {
    client
        .channel_open_direct_tcpip(host_to_connect, port_to_connect, "harness-originator", 0)
        .await
}

/// Opens a `direct-streamlocal` channel — the unix-socket forward
/// helper. The open-failure shape is the `Err` verbatim.
pub async fn open_direct_streamlocal(
    client: &Handle<HarnessClientHandler>,
    socket_path: &str,
) -> Result<Channel<Msg>, Error> {
    client.channel_open_direct_streamlocal(socket_path).await
}

/// Opens an x11 channel — the never-feature open helper (ADR-003 §5).
/// The open-failure shape is the `Err` verbatim.
pub async fn open_x11_channel(
    client: &Handle<HarnessClientHandler>,
    originator_address: &str,
    originator_port: u32,
) -> Result<Channel<Msg>, Error> {
    client
        .channel_open_x11(originator_address, originator_port)
        .await
}

/// Sends an exec request over a channel (`want_reply` set — the
/// success/failure shape rides [`ChannelMsg::Success`]/[`ChannelMsg::Failure`]
/// through [`next_message`]).
pub async fn request_exec(channel: &Channel<Msg>, command: &str) -> Result<(), Error> {
    channel.exec(true, command).await
}

/// Sends a shell request over a channel (`want_reply` set).
pub async fn request_shell(channel: &Channel<Msg>) -> Result<(), Error> {
    channel.request_shell(true).await
}

/// Sends a subsystem request over a channel (`want_reply` set).
pub async fn request_subsystem(channel: &Channel<Msg>, name: &str) -> Result<(), Error> {
    channel.request_subsystem(true, name).await
}

/// Sends a pty request over a channel (`want_reply` set — the pty row
/// rides the standard success/failure observation).
pub async fn request_pty(channel: &Channel<Msg>) -> Result<(), Error> {
    channel.request_pty(true, "xterm", 80, 24, 0, 0, &[]).await
}

/// Sends an env request over a channel (`want_reply` set — the OQ-09
/// unconditional-decline rows' observation).
pub async fn request_env(channel: &Channel<Msg>, name: &str, value: &str) -> Result<(), Error> {
    channel.set_env(true, name, value).await
}

/// Sends an x11-req request over a channel (`want_reply` set — the
/// never-feature request rows).
pub async fn request_x11(channel: &Channel<Msg>) -> Result<(), Error> {
    channel
        .request_x11(true, false, "MIT-MAGIC-COOKIE-1", "harness-cookie", 0)
        .await
}

/// Sends an agent-forwarding request over a channel (`want_reply` set
/// — the never-feature request rows; ADR-003 §5).
pub async fn request_agent_forward(channel: &Channel<Msg>) -> Result<(), Error> {
    channel.agent_forward(true).await
}

/// Requests a `tcpip-forward` listen (`-R` ingress) — the
/// success/failure shape is the result verbatim
/// ([`Error::RequestDenied`] on a declined listen; a `port = 0`
/// request's actually-bound number rides the success value — sshd
/// parity).
pub async fn tcpip_forward(
    client: &Handle<HarnessClientHandler>,
    address: &str,
    port: u32,
) -> Result<u32, Error> {
    client.tcpip_forward(address, port).await
}

/// Requests a `cancel-tcpip-forward`.
pub async fn cancel_tcpip_forward(
    client: &Handle<HarnessClientHandler>,
    address: &str,
    port: u32,
) -> Result<(), Error> {
    client.cancel_tcpip_forward(address, port).await
}

/// Requests a `streamlocal-forward` listen (the unix `-R` ingress).
pub async fn streamlocal_forward(
    client: &Handle<HarnessClientHandler>,
    socket_path: &str,
) -> Result<(), Error> {
    client.streamlocal_forward(socket_path).await
}

/// Requests a `cancel-streamlocal-forward`.
pub async fn cancel_streamlocal_forward(
    client: &Handle<HarnessClientHandler>,
    socket_path: &str,
) -> Result<(), Error> {
    client.cancel_streamlocal_forward(socket_path).await
}

/// Drains the acceptor's event channel — the row's observation shape
/// (`None` once the acceptor's sender went, i.e. after the session's
/// end).
pub async fn next_forwarded_open(
    receiver: &mut tokio::sync::mpsc::UnboundedReceiver<ForwardedOpen>,
) -> Option<ForwardedOpen> {
    receiver.recv().await
}

/// Observes the next channel message — `None` after the channel has
/// closed (russh's EOF-of-conversation shape).
pub async fn next_message(channel: &mut Channel<Msg>) -> Option<ChannelMsg> {
    channel.wait().await
}

/// Awaits the channel's exit status — drains interim messages until
/// [`ChannelMsg::ExitStatus`] (the wave-3 derivation contract's
/// observation shape) or the channel's end, which is `None`.
pub async fn await_exit_status(channel: &mut Channel<Msg>) -> Option<u32> {
    loop {
        match next_message(channel).await {
            Some(ChannelMsg::ExitStatus { exit_status }) => return Some(exit_status),
            Some(_) => continue,
            None => return None,
        }
    }
}

/// Awaits the channel's close — drains interim messages until
/// [`ChannelMsg::Close`] (observed: `true`) or the channel's end
/// (`false`).
pub async fn await_close(channel: &mut Channel<Msg>) -> bool {
    loop {
        match next_message(channel).await {
            Some(ChannelMsg::Close) => return true,
            Some(_) => continue,
            None => return false,
        }
    }
}