//! The client op rows (`w4-op-rows`), `client.md` §op rows in code:
//! the family op → vanilla translation the reverse-runner shape
//! drives (the hub dialed [`SshConnection`](crate::SshConnection) and
//! now opens every channel itself). **Type symmetry at the seam is
//! the contract:** family-protocol types in, vanilla wire ops out, no
//! vanilla-concept leaks up; the rows' inputs are the mapped
//! sibling crates' types (the alktty session shapes for the exec/pty
//! rows, alkgit's `Service` for the git row) and the surfaces are
//! crate-owned (`RemoteSession` / `TunnelChannel` / `GitPackSession`),
//! the failure taxonomy [`OpError`](crate::OpError). All rows share
//! the russh 0.65 mechanics POC-3 §D.3 recorded:
//!
//! - `Channel::exec()`: there is **no `request_exec` at this pin**.
//! - A request's Success/Failure replies arrive **on the channel
//!   queue via `wait()`**, not as `Err` from the request call, so every
//!   request-carrying row drains until the expected message,
//!   `continue`-ing past the interleaved rest, and maps a `Failure`
//!   to [`OpError::RequestRefused`](crate::OpError). The drain is
//!   message-capped (bounded resources: a peer that never answers
//!   cannot loop it unboundedly; the *connection's* wall clock, the
//!   dial's always-set [`Limits::inactivity_timeout`](crate::Limits),
//!   bounds the idle waits).
//! - **The typed opens resolve `Err` directly on a server reject**
//!   (POC-1 §6) to [`OpError::OpenDeclined`](crate::OpError), family-mirrored
//!   [`OpenDeclineReason`](crate::OpenDeclineReason), **and the
//!   connection stays usable after the decline** (row-pinned per
//!   decline class).
//! - **No channel-queue consumption where a `into_stream()` conversion
//!   follows** (POC-1 §6's fresh-`wait()` caveat): the tunnel rows
//!   (the only `into_stream` consumers) go typed-open → straight
//!   conversion, nothing between. The exit-**consuming** rows
//!   (exec/pty, git) cannot ride `into_stream` at all, since the
//!   conversion's reader side swallows every non-`Data` message,
//!   which is exactly where `ExitStatus`/`ExitSignal` live, so they
//!   run the **split-halves wait-pump** (`Channel::split()` + a
//!   spawned task draining `wait()`), the shape the server
//!   direction's git row landed for the same reason
//!   (`w4-git-exit-close`). The exit-ride is the family plane's (§the
//!   session rows) or the consumed-report oneshot's (§the git row).
//! - `window_change`/`signal` are fire-and-forget on the wire (no SSH
//!   reply exists); the pumps send and drop the outcome, never
//!   awaiting one.
//!
//! ## The family plane (the session rows)
//!
//! The exec/pty session's data plane is the **alktty chunk framing in
//! both directions**, the exact plane the server direction's session
//! bridge speaks ([`crate::tty_map`]), consumed not re-derived:
//! stdin chunks (`STREAM_STDIN`) ride the wire `data` (the
//! zero-length sentinel = wire EOF, the bridge's decided translation,
//! the session keeps pumping after it); `CTRL_IN` chunks translate to
//! the wire request rows (`Resize` → `window_change` (the session
//! dimension clamp is not needed; the family plane is already
//! `u16`), `Signal{name}` → `signal` through the shared table's
//! name→`Sig` map, `Eof` → wire EOF); wire `data` → stdout chunks,
//! wire `extended_data(1)` → stderr chunks; and the exit rides
//! **in-band as the family's exit chunk** (`STREAM_CTRL_OUT`,
//! exit-chunk-is-last, alktty's ADR-004, mapping.md's contract):
//! wire `ExitStatus(N)` → code `N`, wire `ExitSignal(name)` →
//! `−signum` through the shared table's inverse read
//! ([`crate::exit_code`], the derivation contract's client arm; a
//! name with no table entry rides the `−1` sentinel, the mapping
//! never invents). A session that ends with **no exit report**
//! delivers no exit chunk; the output half just ends (mapping.md's
//! no-exit-chunk close family: "reports nothing"). The session's
//! ttymodes never ride the wire pty request (the OQ-44
//! reserved-JSON posture mirrored: the door never invents wire
//! terminal modes from family data); a family `term: None` rides the
//! wire's empty-terminal-string value.
//!
//! ## The row inventory
//!
//! | Op | Feature | Vanilla translation | Surface |
//! |---|---|---|---|
//! | pipe-mode exec | `session` | `channel_open_session` + `exec` | [`SshConnection::exec`] |
//! | shell with/without a pty | `session` | `channel_open_session` + `pty_request`? + `shell_request` | [`SshConnection::shell`] |
//! | exec in a pty (`ssh -t host "cmd"`) | `session` | `pty_request` then `exec` | [`SshConnection::pty_exec`] |
//! | tunnel target (`-L`; the alksocks `DialFn`) | `tunnel` | `channel_open_direct_tcpip` | [`SshConnection::tunnel_direct`] |
//! | the streamlocal twin | `tunnel` | `channel_open_direct_streamlocal`, runtime-Unix-gated | [`SshConnection::tunnel_direct_unix`] |
//! | upstream git service | `git` | the shared grammar's command + `exec`; the pack duplex | [`SshConnection::git_service`] |
//! | sftp | *every build* | **none**, the OQ-13 decline placeholder | [`SshConnection::sftp`] |
//!
//! The exec-command **length budget** applies identically to the
//! command-carrying rows (the server direction's pre-parse cap
//! mirrored at the command-construction gate; bounded resources,
//! an unbounded command buffer can never ride a dial):
//! [`OpError::CommandTooLong`](crate::OpError) before any wire op.
//!
//! No client-side in-crate channel cap *admits* ops (the server
//! direction's cap is a wire-battery defense; the client rows are
//! hub-driven (the family app chose the dial): bounded resources on
//! this direction are the config's always-on wall clock/keepalives,
//! the russh windows, the bounded duplexes/mpscs the pumps ride, and
//! the drain cap; every spawn bounded by its own task's end
//! conditions.
//!
//! ## Spawn posture
//!
//! The op methods await their launch (typed open, requests, the
//! reply drains) on the *caller's* task, never the russh packet
//! loop's, so the server direction's spawn-and-return deadlock class
//! does not exist here; the long-lived pumps spawn at launch and end
//! on their terminal conditions (the exit chunk, the channel's end,
//! or the hub dropping the surface; a drop is wire-clean: the input
//! pump's terminal path closes the channel, the pump pair reaping as
//! their duplex halves drop).

#[cfg(any(feature = "session", feature = "git"))]
use russh::ChannelMsg;
#[cfg(any(feature = "session", feature = "git"))]
use russh::client::Msg;
#[cfg(feature = "git")]
use tokio::io::AsyncWriteExt;

use crate::client::SshConnection;
#[cfg(feature = "session")]
use crate::client_handler::ClientHandler;
use crate::error::OpError;

/// The op-row pump duplex capacity, the crate's bridge-pair
/// convention (the wave-2 loopback harness's per-direction capacity),
/// keeping every pump's backpressure bounded.
#[cfg(any(feature = "session", feature = "git"))]
const OP_DUPLEX_CAPACITY: usize = 65_536;

/// The pump-side read buffer, the SSH-side granularity (per-chunk
/// payloads are bounded by the dial's `maximum_packet_size`).
#[cfg(feature = "git")]
const PUMP_BUF: usize = 16 * 1024;

/// The request-reply drain's message cap (bounded resources, the
/// §module docs' drain row): a peer answering only with non-reply
/// messages cannot loop the drain unboundedly.
#[cfg(any(feature = "session", feature = "git"))]
const DRAIN_MESSAGE_CAP: usize = 1024;

/// The originator label the door puts on the wire's `direct-tcpip`
/// opens (the SSH originator fields are what a vanilla client fills
/// in for its own local listener; this door's hub-driven dials carry
/// the door's label and no port, recorded decision: nothing
/// invented, nothing leaked, the peer's logs name the door).
#[cfg(feature = "tunnel")]
const TUNNEL_ORIGINATOR: &str = "alk-ssh";

/// The wire exit status → the family `i32` exit-code convention: a
/// code beyond the family `i32` range is unwritable as an exit-chunk
/// code; the clamp rides (defensive corner, vanilla command codes
/// are far below it; the mapping never invents a value, it narrows).
#[cfg(any(feature = "session", feature = "git"))]
fn family_exit_code(status: u32) -> i32 {
    status.min(i32::MAX as u32) as i32
}

/// The typed-open decline's translation point: the russh error → the
/// op taxonomy (a channel-open failure →
/// [`OpError::OpenDeclined`](crate::OpError::OpenDeclined),
/// family-mirrored; everything else transparent). The russh level's
/// test observable (`POC-1 §6`'s `Err(ChannelOpenFailure)`, the
/// awaiting typed call's resolution) rides this point; the
/// connection stays usable after every decline (row-pinned).
#[cfg(any(feature = "session", feature = "git", feature = "tunnel"))]
fn open_error(error: russh::Error) -> OpError {
    OpError::from(error)
}

/// The request-reply drain (the §module docs' recorded mechanics,
/// shared by the exec/pty rows and the git row): Success/Failure via
/// `wait()`, never an `Err` from the request call; `continue` past
/// the interleaved rest (the POC-3 §D.3 break-on-first-message trap,
/// not repeated), message-capped. `Failure` is the wire
/// request-refusal class ([`OpError::RequestRefused`]); the
/// channel's end ahead of any reply is [`OpError::ChannelEnded`].
#[cfg(any(feature = "session", feature = "git"))]
async fn drain_request_reply(channel: &mut russh::Channel<Msg>) -> Result<(), OpError> {
    for _ in 0..DRAIN_MESSAGE_CAP {
        match channel.wait().await {
            Some(ChannelMsg::Success) => return Ok(()),
            Some(ChannelMsg::Failure) => return Err(OpError::RequestRefused),
            Some(ChannelMsg::Close) => return Err(OpError::ChannelEnded),
            Some(_) => continue,
            None => return Err(OpError::ChannelEnded),
        }
    }
    Err(OpError::ChannelEnded)
}

/// Drains one request's reply; on any refusal, closes the channel
/// behind it (the row owned the channel; the launch never
/// half-serves) and returns the error. The launch rows' shared step.
#[cfg(any(feature = "session", feature = "git"))]
async fn drain_launch_request(channel: &mut russh::Channel<Msg>) -> Result<(), OpError> {
    let outcome = drain_request_reply(channel).await;
    if outcome.is_err() {
        let _ = channel.close().await;
    }
    outcome
}

// ---------------------------------------------------------------------------
// The sftp placeholder, every build (OQ-13; no feature, no dep)
// ---------------------------------------------------------------------------

impl SshConnection {
    /// The sftp op, the client-side half of the OQ-13 decline
    /// placeholder: **declines in every build, before any wire op**
    /// (never a half-opened subsystem channel). The feature exists
    /// and ships in the build string; the route does not, and arrives
    /// with alkfs as its own ADR; both directions' placeholders are
    /// then replaced by real rows (`client.md` §What the crate does
    /// *not* include). Compiles feature-free; the `Ok` type is
    /// `Infallible` by design; nothing can ever populate it here.
    pub async fn sftp(&self) -> Result<std::convert::Infallible, OpError> {
        Err(OpError::SftpNotServed)
    }
}

// ---------------------------------------------------------------------------
// The session rows: exec / pty (under `session`)
// ---------------------------------------------------------------------------

/// A remote **session**: the family session plane over the runner
/// (the session rows' surface): the alktty chunk framing in both
/// directions, exactly the plane the server direction's session
/// bridge speaks (`tty_map`), reversed. Two duplex halves
/// the hub drives; the pumps are spawned at the op's launch (§the
/// family plane in the module docs for the full chunk map). Dropping
/// the session (both halves) is wire-clean: the input pump's
/// terminal path closes the channel.
#[cfg(feature = "session")]
pub struct RemoteSession {
    /// The hub→wire chunk plane: stdin chunks + `CTRL_IN` controls
    /// (`alktty::wire::ChunkWriter` framing; the zero-length stdin
    /// chunk = wire EOF; `Resize`/`Signal`/`Eof` controls = the wire
    /// request rows).
    pub input: tokio::io::DuplexStream,
    /// The wire→hub chunk plane: stdout/stderr chunks + the exit
    /// chunk in-band, last (exit-chunk-is-last).
    pub output: tokio::io::DuplexStream,
}

#[cfg(feature = "session")]
impl std::fmt::Debug for RemoteSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RemoteSession").finish_non_exhaustive()
    }
}

/// Which request launches a session: the family op's payload
/// (§the row inventory; pipe-mode exec vs the shell form).
#[cfg(feature = "session")]
enum LaunchRequest {
    Exec(String),
    Shell,
}

#[cfg(feature = "session")]
impl SshConnection {
    /// The pipe-mode exec row: `ssh host "cmd"` in, family-typed;
    /// the session channel's open + the wire `exec` request, then the
    /// chunk-framed session (§the family plane; stderr separated).
    /// The command rides the wire verbatim (the remote end's shell
    /// parses it, the vanilla exec grammar; never argv-split).
    pub async fn exec(&self, command: impl Into<String>) -> Result<RemoteSession, OpError> {
        let command = command.into();
        if command.len() > self.budgets().max_exec_command_len {
            return Err(OpError::CommandTooLong);
        }
        open_remote_session(self.russh_handle(), None, LaunchRequest::Exec(command)).await
    }

    /// The pty-mode exec row (`ssh -t host "cmd"`): the pty request
    /// carries the family
    /// [`alktty::negotiation::TerminalParamsWire`](alktty::negotiation::
    /// TerminalParamsWire) first (its acceptance drained), then the
    /// exec. **A refused pty refuses the whole row**: never a
    /// half-capability, never a silent pty-less fall.
    pub async fn pty_exec(
        &self,
        params: alktty::negotiation::TerminalParamsWire,
        command: impl Into<String>,
    ) -> Result<RemoteSession, OpError> {
        let command = command.into();
        if command.len() > self.budgets().max_exec_command_len {
            return Err(OpError::CommandTooLong);
        }
        open_remote_session(
            self.russh_handle(),
            Some(params),
            LaunchRequest::Exec(command),
        )
        .await
    }

    /// The shell row ("own a shell on the runner"): with
    /// `params`, this is the pty mode (the interactive `ssh` shape); with
    /// `None`, the pipe-mode shell (the `ssh -T` shape; a pty-less
    /// shell request is legitimate vanilla and served, never declined,
    /// mirroring the server direction's decided parity).
    pub async fn shell(
        &self,
        params: Option<alktty::negotiation::TerminalParamsWire>,
    ) -> Result<RemoteSession, OpError> {
        open_remote_session(self.russh_handle(), params, LaunchRequest::Shell).await
    }
}

/// The session rows' launcher (§module docs): typed open → the pty? →
/// the launch request (each request-reply drained; a refusal closes
/// the channel behind itself) → the split-halves pumps.
#[cfg(feature = "session")]
async fn open_remote_session(
    handle: &russh::client::Handle<ClientHandler>,
    pty: Option<alktty::negotiation::TerminalParamsWire>,
    request: LaunchRequest,
) -> Result<RemoteSession, OpError> {
    let mut channel = handle.channel_open_session().await.map_err(open_error)?;

    if let Some(params) = &pty {
        channel
            .request_pty(
                true,
                params.term.as_deref().unwrap_or_default(),
                u32::from(params.cols),
                u32::from(params.rows),
                u32::from(params.pixel_width),
                u32::from(params.pixel_height),
                &[],
            )
            .await
            .map_err(open_error)?;
        drain_launch_request(&mut channel).await?;
    }

    match request {
        LaunchRequest::Exec(command) => {
            channel
                .exec(true, command.into_bytes())
                .await
                .map_err(open_error)?;
        }
        LaunchRequest::Shell => {
            channel.request_shell(true).await.map_err(open_error)?;
        }
    }
    drain_launch_request(&mut channel).await?;

    let (read_half, write_half) = channel.split();
    let (kill, kill_rx) = tokio::sync::watch::channel(false);
    let (in_tx, in_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY);
    let (out_tx, out_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY);
    tokio::spawn(session_input_pump(write_half, in_rx, kill_rx));
    tokio::spawn(session_output_pump(read_half, out_tx, kill));
    Ok(RemoteSession {
        input: in_tx,
        output: out_rx,
    })
}

/// The session row's **input pump**: the hub's chunk plane → the
/// wire (§the family plane). The zero-length stdin sentinel / a
/// `CTRL_IN` `Eof` / the hub's input half dropping = the wire EOF
/// *once*, then the idle-for-kill state (the session keeps pumping);
/// the terminal kill (from the output pump's terminal paths) → the
/// wire `channel_close`; the session's drop is wire-clean.
#[cfg(feature = "session")]
async fn session_input_pump(
    write_half: russh::ChannelWriteHalf<Msg>,
    input: tokio::io::DuplexStream,
    mut kill: tokio::sync::watch::Receiver<bool>,
) {
    let mut done = false;
    let mut reader = alktty::wire::ChunkReader::new(input);
    loop {
        tokio::select! {
            frame = hub_frame(&mut reader), if !done => {
                match frame {
                    HubFrame::Chunk(chunk) => match chunk.stream_type {
                        alktty::wire::STREAM_STDIN => {
                            if chunk.bytes.is_empty() {
                                let _ = write_half.eof().await;
                                done = true;
                            } else if write_half.data_bytes(chunk.bytes).await.is_err() {
                                return;
                            }
                        }
                        alktty::wire::STREAM_CTRL_IN => {
                            match alktty::ControlMessage::from_slice(&chunk.bytes) {
                                Ok(alktty::ControlMessage::Resize {
                                    cols,
                                    rows,
                                    pixel_width,
                                    pixel_height,
                                }) => {
                                    let _ = write_half
                                        .window_change(
                                            u32::from(cols),
                                            u32::from(rows),
                                            u32::from(pixel_width),
                                            u32::from(pixel_height),
                                        )
                                        .await;
                                }
                                Ok(alktty::ControlMessage::Signal { name }) => {
                                    let _ = write_half
                                        .signal(crate::exit_code::sig_from_name(&name))
                                        .await;
                                }
                                Ok(alktty::ControlMessage::Eof) => {
                                    let _ = write_half.eof().await;
                                    done = true;
                                }
                                _ => {}
                            }
                        }
                        _ => {}
                    },
                    HubFrame::Break => {
                        let _ = write_half.close().await;
                        return;
                    }
                    HubFrame::Closed => {
                        let _ = write_half.eof().await;
                        done = true;
                    }
                }
            }
            _ = kill.changed() => {
                let _ = write_half.close().await;
                return;
            }
        }
    }
}

/// One frame read from the hub→door plane, in the *no-disambiguation*
/// read direction: this plane's chunks are the family protocol's own
/// (stdin `0` / `CTRL_IN` `3`); no negotiation-error frames exist
/// toward the door, so the read is the protocol's direct
/// [`ChunkReader::read_chunk`] (a `0x00`-leading frame here is a
/// zero-length stdin chunk (the sentinel) and reads as exactly
/// that). A rogue frame is the bounds-checked [`RawError`], mapped to
/// the pump's [`HubFrame::Break`].
#[cfg(feature = "session")]
enum HubFrame {
    Chunk(alktty::wire::Chunk),
    Closed,
    Break,
}

#[cfg(feature = "session")]
async fn hub_frame(reader: &mut alktty::wire::ChunkReader<tokio::io::DuplexStream>) -> HubFrame {
    match reader.read_chunk().await {
        Ok(chunk) => HubFrame::Chunk(chunk),
        Err(alktty::wire::RawError::ConnectionClosed) => HubFrame::Closed,
        Err(_) => HubFrame::Break,
    }
}

/// The session row's **output pump**: the wire → the hub's chunk
/// plane (§the family plane). `data` → stdout chunks,
/// `extended_data(1)` → stderr chunks, and the `ExitStatus` /
/// `ExitSignal` terminal → the in-band exit chunk (last), then the
/// kill fires and the input pump closes the wire. A channel end with
/// no report → no exit chunk; the output half just ends.
#[cfg(feature = "session")]
async fn session_output_pump(
    mut read_half: russh::ChannelReadHalf,
    output: tokio::io::DuplexStream,
    kill: tokio::sync::watch::Sender<bool>,
) {
    let mut writer = alktty::wire::ChunkWriter::new(output);
    loop {
        match read_half.wait().await {
            None | Some(ChannelMsg::Close) => {
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::Data { data }) => {
                let chunk = alktty::wire::Chunk {
                    stream_type: alktty::wire::STREAM_STDOUT,
                    bytes: data,
                };
                if writer.write_chunk(&chunk).await.is_err() {
                    let _ = kill.send(true);
                    return;
                }
            }
            Some(ChannelMsg::ExtendedData { data, ext: 1 }) => {
                let chunk = alktty::wire::Chunk {
                    stream_type: alktty::wire::STREAM_STDERR,
                    bytes: data,
                };
                if writer.write_chunk(&chunk).await.is_err() {
                    let _ = kill.send(true);
                    return;
                }
            }
            Some(ChannelMsg::ExitStatus { exit_status }) => {
                write_exit_chunk(&mut writer, family_exit_code(exit_status)).await;
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::ExitSignal { signal_name, .. }) => {
                let name = crate::exit_code::sig_control_name(&signal_name);
                write_exit_chunk(&mut writer, crate::exit_code::signal_code_from_name(&name)).await;
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::Eof) => continue,
            Some(_) => {}
        }
    }
}

/// The family exit chunk's write: `{"type":"exit","code":N}` as the
/// in-band terminal chunk (exit-chunk-is-last), serialized by the
/// family currency's own serializer. An unserializable exit (not a
/// reachable shape) writes nothing; the no-report family, never a
/// corrupted frame.
#[cfg(feature = "session")]
async fn write_exit_chunk(
    writer: &mut alktty::wire::ChunkWriter<tokio::io::DuplexStream>,
    code: i32,
) {
    if let Ok(bytes) = (alktty::ControlMessage::Exit { code }).to_json() {
        let _ = writer
            .write_chunk(&alktty::wire::Chunk {
                stream_type: alktty::wire::STREAM_CTRL_OUT,
                bytes,
            })
            .await;
    }
}

// ---------------------------------------------------------------------------
// The tunnel rows: direct-tcpip / direct-streamlocal (under `tunnel`)
// ---------------------------------------------------------------------------

/// A tunnel channel, the tunnel rows' surface: a plain byte duplex
/// over the open channel (`-L`'s pump; the alksocks `DialFn` composes
/// exactly this type). Nothing vanilla-typed rides the family
/// surface; the duplex's `poll_shutdown` carries the wire EOF.
#[cfg(feature = "tunnel")]
pub struct TunnelChannel {
    /// The russh channel's byte duplex (`pub(crate)`; the `-R` leg's
    /// acceptor payloads construct the type over a server-initiated
    /// open's `into_stream()` conversion; downstream crates see the
    /// type opaque as ever).
    pub(crate) stream: russh::ChannelStream<russh::client::Msg>,
}

#[cfg(feature = "tunnel")]
impl std::fmt::Debug for TunnelChannel {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TunnelChannel").finish_non_exhaustive()
    }
}

#[cfg(feature = "tunnel")]
impl tokio::io::AsyncRead for TunnelChannel {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.stream).poll_read(cx, buf)
    }
}

#[cfg(feature = "tunnel")]
impl tokio::io::AsyncWrite for TunnelChannel {
    fn poll_write(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        std::pin::Pin::new(&mut self.stream).poll_write(cx, buf)
    }

    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.stream).poll_flush(cx)
    }

    fn poll_shutdown(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.stream).poll_shutdown(cx)
    }
}

#[cfg(feature = "tunnel")]
impl SshConnection {
    /// The tunnel-target row (`-L`; the alksocks `DialFn` composes
    /// over it): the typed open; server reject →
    /// [`OpError::OpenDeclined`](crate::OpError) with the mirrored
    /// class, and the connection stays usable (row-pinned per
    /// decline class). Between the open and the `into_stream()`
    /// conversion **nothing else happens**: no channel-queue
    /// consumption (POC-1 §6's caveat, respected by construction; the
    /// data pump is the raw byte duplex).
    pub async fn tunnel_direct(
        &self,
        host: impl Into<String>,
        port: u32,
    ) -> Result<TunnelChannel, OpError> {
        let channel = self
            .russh_handle()
            .channel_open_direct_tcpip(host, port, TUNNEL_ORIGINATOR, 0)
            .await
            .map_err(open_error)?;
        Ok(TunnelChannel {
            stream: channel.into_stream(),
        })
    }

    /// The streamlocal twin (`direct-streamlocal@openssh.com`): same
    /// typed-open/pump shape, the OQ-11 runtime-Unix gate mirrored;
    /// **compiles wherever `tunnel` compiles**, and off-Unix the op
    /// declines cleanly ([`OpError::UnixUnavailable`](crate::OpError))
    /// before any wire op.
    pub async fn tunnel_direct_unix(
        &self,
        socket_path: impl Into<String>,
    ) -> Result<TunnelChannel, OpError> {
        #[cfg(not(unix))]
        {
            let _ = socket_path;
            Err(OpError::UnixUnavailable)
        }
        #[cfg(unix)]
        {
            let channel = self
                .russh_handle()
                .channel_open_direct_streamlocal(socket_path)
                .await
                .map_err(open_error)?;
            Ok(TunnelChannel {
                stream: channel.into_stream(),
            })
        }
    }
}

// ---------------------------------------------------------------------------
// The git row: upstream exec consumption (under `git`)
// ---------------------------------------------------------------------------

/// An upstream **git service channel**, the git row's surface: the
/// raw pack duplex (the git protocol's own byte plane rides the
/// channel untouched), the sshd-side stderr split out, and the
/// consumed exit report at the family's `i32` exit-code convention
/// (`ExitStatus(N)` → `N`; `ExitSignal` → `−signum` through the
/// shared table, the mapping never invents; a receive `Err` on
/// `exit` = the pump ended with **no exit report at all**, the
/// no-exit-chunk close family's mirror).
#[cfg(feature = "git")]
pub struct GitPackSession {
    /// The pack bytes the hub pushes to the wire (request lines,
    /// haves, the pushed pack). Ending the half = wire EOF.
    pub input: tokio::io::DuplexStream,
    /// The wire `data` → the pack bytes (the advertisement, acks,
    /// the pack).
    pub output: tokio::io::DuplexStream,
    /// The wire `extended_data(1)` (the remote program's stderr
    /// diagnostics), raw bytes; a hub uninterested in diagnostics
    /// drops the half (the pump then discards further stderr).
    pub stderr: tokio::io::DuplexStream,
    /// The consumed exit report (see the type docs).
    pub exit: tokio::sync::oneshot::Receiver<i32>,
}

#[cfg(feature = "git")]
impl std::fmt::Debug for GitPackSession {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GitPackSession").finish_non_exhaustive()
    }
}

#[cfg(feature = "git")]
impl SshConnection {
    /// The upstream git row: the shared exec grammar's command
    /// (`git-{upload,receive}-pack '/<repo>'`, built through
    /// `crate::dispatch`'s construction arm, the parse arm's exact
    /// inverse; no forked grammar), pointed at the vanilla `sshd`:
    /// typed open + `exec`, the pack duplex pumped, the exit report
    /// consumed.
    ///
    /// **The pre-wire gates, riding the shared grammar's verdicts:**
    /// the repo id validates against the ADR-008 grammar (the git
    /// row's server-direction gate mirrored, `crate::git_map`) →
    /// [`OpError::InvalidRepoId`](crate::OpError);
    /// the constructed command is checked against the exec-length
    /// budget ([`crate::Limits::max_exec_command_len`]) →
    /// [`OpError::CommandTooLong`](crate::OpError), both **before any
    /// wire op**. (The grammar's other decline classes,
    /// `unknown-service` / `service-only`, are structurally
    /// unreachable here: the [`alkgit::Service`] input type carries
    /// only the two serving services; the construction arm can emit
    /// no other token.)
    pub async fn git_service(
        &self,
        service: alkgit::Service,
        repo_id: &str,
    ) -> Result<GitPackSession, OpError> {
        let route = match service {
            alkgit::Service::UploadPack => crate::dispatch::GitServiceRoute::UploadPack,
            alkgit::Service::ReceivePack => crate::dispatch::GitServiceRoute::ReceivePack,
        };
        if !crate::git_map::repo_id_validates(repo_id) {
            return Err(OpError::InvalidRepoId);
        }
        let command = crate::dispatch::build_git_exec_command(route, repo_id);
        if command.len() > self.budgets().max_exec_command_len {
            return Err(OpError::CommandTooLong);
        }

        let mut channel = self
            .russh_handle()
            .channel_open_session()
            .await
            .map_err(open_error)?;
        channel
            .exec(true, command.into_bytes())
            .await
            .map_err(open_error)?;
        drain_launch_request(&mut channel).await?;

        let (read_half, write_half) = channel.split();
        let (kill, kill_rx) = tokio::sync::watch::channel(false);
        let (in_tx, in_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY);
        let (out_tx, out_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY);
        let (err_tx, err_rx) = tokio::io::duplex(OP_DUPLEX_CAPACITY);
        let (exit_tx, exit_rx) = tokio::sync::oneshot::channel();
        tokio::spawn(git_input_pump(write_half, in_rx, kill_rx));
        tokio::spawn(git_output_pump(read_half, out_tx, err_tx, exit_tx, kill));
        Ok(GitPackSession {
            input: in_tx,
            output: out_rx,
            stderr: err_rx,
            exit: exit_rx,
        })
    }
}

/// The git row's **raw input pump**: the hub's pack bytes → the wire;
/// the input half's end (or any read error) = the wire EOF *once*,
/// then the idle-for-kill state (the advertisement/acks/exit keep
/// flowing); the kill (the output pump's terminal paths) → the wire
/// `channel_close`.
#[cfg(feature = "git")]
async fn git_input_pump(
    write_half: russh::ChannelWriteHalf<Msg>,
    mut input: tokio::io::DuplexStream,
    mut kill: tokio::sync::watch::Receiver<bool>,
) {
    let mut done = false;
    let mut buf = vec![0u8; PUMP_BUF];
    loop {
        if done {
            if kill.changed().await.is_err() {
                let _ = write_half.close().await;
                return;
            }
            let _ = write_half.close().await;
            return;
        }
        tokio::select! {
            n = tokio::io::AsyncReadExt::read(&mut input, &mut buf) => {
                match n {
                    Ok(0) | Err(_) => {
                        let _ = write_half.eof().await;
                        done = true;
                    }
                    Ok(count) => {
                        if write_half.data(&buf[..count]).await.is_err()
                        {
                            return;
                        }
                    }
                }
            }
            _ = kill.changed() => {
                let _ = write_half.close().await;
                return;
            }
        }
    }
}

/// The git row's **output pump**: the wire → the pack duplex +
/// stderr; the exit report consumed here (the family code) and the
/// kill fired, then the input pump closes the wire. A channel end
/// with no report → no exit fired; the `exit` oneshot drops (the
/// hub's receive reads the no-report mirror).
#[cfg(feature = "git")]
async fn git_output_pump(
    mut read_half: russh::ChannelReadHalf,
    mut output: tokio::io::DuplexStream,
    mut stderr: tokio::io::DuplexStream,
    exit: tokio::sync::oneshot::Sender<i32>,
    kill: tokio::sync::watch::Sender<bool>,
) {
    let mut stderr_gone = false;
    loop {
        match read_half.wait().await {
            None | Some(ChannelMsg::Close) => {
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::Data { data }) => {
                if output.write_all(&data).await.is_err() {
                    let _ = kill.send(true);
                    return;
                }
            }
            Some(ChannelMsg::ExtendedData { data, ext: 1 }) => {
                if !stderr_gone && stderr.write_all(&data).await.is_err() {
                    stderr_gone = true;
                }
            }
            Some(ChannelMsg::ExitStatus { exit_status }) => {
                let _ = exit.send(family_exit_code(exit_status));
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::ExitSignal { signal_name, .. }) => {
                let name = crate::exit_code::sig_control_name(&signal_name);
                let _ = exit.send(crate::exit_code::signal_code_from_name(&name));
                let _ = kill.send(true);
                return;
            }
            Some(ChannelMsg::Eof) => continue,
            Some(_) => {}
        }
    }
}

#[cfg(all(test, any(feature = "session", feature = "git", feature = "tunnel")))]
mod tests {
    #[cfg(any(feature = "session", feature = "git"))]
    use super::family_exit_code;
    use super::open_error;
    use crate::error::OpError;
    use russh::ChannelOpenFailure;

    /// - *Contract stamp:* `client.md` §op rows (no vanilla-concept
    ///   leaks up) + POC-1 §6: the typed open's decline mapping
    ///   preserves the wire class (`error.rs`'s mirror rows; this
    ///   pins the op rows' own consumption point).
    #[test]
    fn typed_open_decline_maps_through_the_mirror() {
        let mapped = open_error(russh::Error::ChannelOpenFailure(
            ChannelOpenFailure::ConnectFailed,
        ));
        assert!(matches!(
            mapped,
            OpError::OpenDeclined(crate::error::OpenDeclineReason::ConnectFailed)
        ));
        let mapped = open_error(russh::Error::ChannelOpenFailure(
            ChannelOpenFailure::AdministrativelyProhibited,
        ));
        assert!(matches!(
            mapped,
            OpError::OpenDeclined(crate::error::OpenDeclineReason::Prohibited)
        ));
    }

    /// - *Contract stamp:* the family exit-currency clamp (§the
    ///   family plane); ordinary codes ride identity, beyond-range
    ///   ones narrow (never invent).
    #[cfg(any(feature = "session", feature = "git"))]
    #[test]
    fn family_exit_code_narrows_beyond_the_i32_range() {
        assert_eq!(family_exit_code(0), 0);
        assert_eq!(family_exit_code(255), 255);
        assert_eq!(family_exit_code(u32::MAX), i32::MAX);
    }
}