//! The op rows' contract rows (`w4-op-rows`) — driven through the
//! in-process fake `sshd` grown with the op-row script
//! ([`sshd_double`](common::sshd_double), the client-direction
//! vehicle): featured + working rows (exec exit-status, pty
//! window-change riding the live pump, the direct-tcpip pump, the git
//! exec round trip), the decline rows (server-rejected opens, the
//! sftp placeholder, the off-Unix streamlocal twin — OQ-11), and the
//! connection-stays-usable row after each decline class. The
//! row/stamp discipline (`tests/common/mod.rs`'s doc) applies
//! verbatim.
//!
//! **Smoke posture** (the task's decided + recorded item): the session
//! rows' cheap real-stack smoke is the alkssh-client-op-row ↔
//! alkssh-server-direction loopback (both russh stacks; the client
//! exec row driven against the server direction's *real* session rows
//! — the chunk bridge's two arms meeting over one live wire), gated
//! on tooling like the wave-3 smokes — skipped-with-reason when
//! absent, never silently green. A real-vanilla-`sshd` dial is wave-5
//! e2e territory explicitly (see the task's Notes).

mod common;

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

use alkssh::{DialConfig, Limits, OpError, SshConnection, dial};

#[cfg(any(feature = "session", feature = "tunnel"))]
use alkssh::OpenDeclineReason;
#[cfg(feature = "session")]
use alkssh::fingerprint;

#[cfg(feature = "tunnel")]
use common::sshd_double::DblDirect;
#[cfg(any(feature = "session", feature = "git"))]
use common::sshd_double::DblExit;
use common::sshd_double::{SshdDouble, SshdObserved, SshdScript, spawn_scripted};
use common::{MemHostKeyPolicy, TestVault, fixture};

/// A vault-backed key source over the alpha fixture — the dial
/// posture (the fixture PEM sealed, decrypt-at-dial).
fn alpha_key_source() -> Arc<dyn alkssh::KeySource> {
    let vault = TestVault::unlocked();
    let blob = vault.seal(fixture("alpha"));
    Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob)))
}

/// Dials a fresh double under the given script — one dial per row per
/// double. Returns the connection, the double (its outcome channel
/// for the clean-close rows) and the double's observations.
async fn dial_scripted(script: SshdScript) -> (SshConnection, SshdDouble, SshdObserved) {
    dial_scripted_limits(script, Limits::default()).await
}

/// [`dial_scripted`] tuned with an explicit budget (the client-side
/// caps are [`Limits`], carried by the dial).
async fn dial_scripted_limits(
    script: SshdScript,
    limits: Limits,
) -> (SshConnection, SshdDouble, SshdObserved) {
    let (double, accepts, observed) =
        spawn_scripted(vec![fixture("bravo").private_key()], true, script).await;
    accepts.register(fixture("alpha").fingerprint);

    let policy = Arc::new(MemHostKeyPolicy::new());
    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
    let mut config = DialConfig::new(
        double.address.clone(),
        "runner",
        alpha_key_source(),
        Arc::clone(&policy) as Arc<dyn alkssh::HostKeyPolicy>,
    );
    config.limits = limits;
    let connection = dial(config).await.expect("the dial succeeds");
    (connection, double, observed)
}

/// The clean-close observable: the explicit close leaves the double's
/// session `Ok` — the decline rows' "the connection stays usable"
/// final act.
async fn expect_clean_close(connection: SshConnection, mut double: SshdDouble) {
    assert!(!connection.is_closed());
    connection.close().await;
    let outcome = double.next_session_outcome().await;
    assert!(
        matches!(outcome, Some(Ok(()))),
        "the clean close is the double's clean session end, got {outcome:?}"
    );
    double.shutdown();
}

/// `next_opened_session` bounded for the "no channel ever opened"
/// assertions (an unbounded `recv` would hang a wrong row).
async fn no_open_observed(double: &mut SshdDouble, within: &str) {
    let deadline = tokio::time::Instant::now() + Duration::from_millis(400);
    let opened = tokio::time::timeout_at(deadline, double.next_opened_session()).await;
    assert!(
        matches!(opened, Ok(None) | Err(_)),
        "no session channel may open before {within}: got {opened:?}"
    );
}

/// OQ-13's client mirror: the sftp op declines cleanly in **every**
/// build (the `sftp` feature ships in the build string but gates no
/// route — the featured-build variant below pins that half too,
/// `--all-features` running both).
///
/// - *Contract stamp:* OQ-13 (the decline placeholder is the shipped
///   behavior, both directions) + `client.md` §op rows (row 5) + §What
///   the crate does *not* include — before any wire op: no session
///   channel ever opens at the peer, no subsystem request ever rides,
///   and the connection stays usable (the clean close is the double's
///   clean session end).
#[tokio::test]
async fn sftp_placeholder_declines_in_every_build_before_any_wire_op() {
    let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await;

    let err = connection
        .sftp()
        .await
        .expect_err("the OQ-13 placeholder declines the sftp op");
    assert!(matches!(err, OpError::SftpNotServed), "got {err:?}");
    assert!(
        observed.execs().is_empty(),
        "the placeholder consulted no wire op"
    );
    no_open_observed(&mut double, "the sftp placeholder's decline").await;
    expect_clean_close(connection, double).await;
}

#[cfg(feature = "session")]
#[tokio::test]
async fn sftp_feature_on_the_placeholder_still_declines() {
    let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await;
    let err = connection
        .sftp()
        .await
        .expect_err("the feature never serves the route (OQ-13 — the placeholder ships)");
    assert!(matches!(err, OpError::SftpNotServed), "got {err:?}");
    assert!(observed.execs().is_empty());
    no_open_observed(&mut double, "the featured placeholder's decline").await;
    expect_clean_close(connection, double).await;
}

#[cfg(feature = "session")]
mod session_rows {
    use super::*;
    use alktty::wire::{ChunkReader, STREAM_CTRL_IN, STREAM_STDIN};
    use alktty::{ControlMessage, negotiation::TerminalParamsWire};
    use tokio::io::AsyncWriteExt;

    /// Frames one chunk by hand (`stream_type` byte + the big-endian
    /// `u32` length + payload — the published header format, read
    /// back only through alktty's own [`ChunkReader`]).
    fn frame(stream_type: u8, payload: &[u8]) -> Vec<u8> {
        let mut out = vec![stream_type];
        out.extend_from_slice(&(payload.len() as u32).to_be_bytes());
        out.extend_from_slice(payload);
        out
    }

    /// Frames one control chunk — the family currency's own
    /// serializer builds the payload.
    async fn control_frame(message: &ControlMessage) -> Vec<u8> {
        let payload = message.to_json().expect("the control serializes");
        frame(STREAM_CTRL_IN, &payload)
    }

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

    /// The exit row: the sentinel rides, the pump delivers.
    async fn send_stdin_eof(session: &mut alkssh::RemoteSession) {
        session
            .input
            .write_all(&frame(STREAM_STDIN, &[]))
            .await
            .expect("the stdin sentinel writes");
    }

    /// - *Contract stamp:* `client.md` §op rows (row 1, the exec
    ///   translation + the family plane) — the wire carries the
    ///   family command verbatim (`exec()`, never `request_exec`);
    ///   stdout/stderr come back as distinct chunks and the exit
    ///   status rides the family chunk in-band, last.
    #[tokio::test]
    async fn exec_row_pumps_the_program_and_the_exit_through_the_family_plane() {
        let script = SshdScript {
            stdout: b"out-ok",
            stderr: b"err-why",
            exit: Some(DblExit::Status(0)),
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let mut session = connection
            .exec("echo out-ok")
            .await
            .expect("the exec row launches");

        assert_eq!(
            observed.execs(),
            vec![b"echo out-ok".to_vec()],
            "the wire carries the family command, never altered"
        );
        send_stdin_eof(&mut session).await;
        let (stdout, stderr, exit) = read_to_exit(session).await;
        assert_eq!(stdout, b"out-ok");
        assert_eq!(stderr, b"err-why");
        assert_eq!(
            exit,
            Some(0),
            "exit-status(0) rode the wire; the family chunk mirrors it in-band"
        );
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* POC-3 §D.3 (the Success/Failure semantics —
    ///   **the Failure reply surfaces on the channel queue's drain,
    ///   not as `Err` from the request call**) →
    ///   [`OpError::RequestRefused`]; the refused row closes its
    ///   channel behind the refusal and the connection stays usable
    ///   (the clean-close observable).
    #[tokio::test]
    async fn refused_exec_reply_surfaces_the_request_refusal() {
        let script = SshdScript {
            request_success: false,
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let err = connection
            .exec("echo refused")
            .await
            .expect_err("the scripted refusal surfaces");
        assert!(matches!(err, OpError::RequestRefused), "got {err:?}");
        assert_eq!(
            observed.execs(),
            vec![b"echo refused".to_vec()],
            "the wire carried the request (its Failure reply is the class)"
        );
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the command-carrying rows' length budget
    ///   (the server direction's pre-parse cap, `Limits::
    ///   max_exec_command_len`, mirrored at the command-construction
    ///   gate): a beyond-budget command declines **before any wire
    ///   op** — no channel, no request, bounded by construction.
    #[tokio::test]
    async fn oversized_exec_declines_before_any_wire_op() {
        let limits = Limits {
            max_exec_command_len: 4,
            ..Limits::default()
        };
        let script = SshdScript {
            stdout: b"never",
            exit: Some(DblExit::Status(0)),
            ..SshdScript::default()
        };
        let (connection, mut double, observed) = dial_scripted_limits(script, limits).await;
        let err = connection
            .exec("echo too long")
            .await
            .expect_err("the budget gate declines before the wire");
        assert!(matches!(err, OpError::CommandTooLong), "got {err:?}");
        assert!(observed.execs().is_empty());
        no_open_observed(&mut double, "the budget gate's decline").await;
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* `client.md` §op rows (row 1, the pty/shell
    ///   translation) — the family [`TerminalParamsWire`] rides the
    ///   wire `pty_request` verbatim (the reserved modes never ride:
    ///   the double sees the params, and the wire request carries the
    ///   empty terminal-modes set), and the shell request launches
    ///   the session.
    #[tokio::test]
    async fn pty_shell_translates_the_family_params_onto_the_wire() {
        let (connection, double, observed) = dial_scripted(SshdScript::default()).await;
        let params = TerminalParamsWire {
            term: Some("xterm-256color".to_owned()),
            cols: 80,
            rows: 24,
            pixel_width: 10,
            pixel_height: 20,
            modes: serde_json::json!({}),
        };
        let session = connection
            .shell(Some(params))
            .await
            .expect("the pty shell row launches");
        assert_eq!(
            observed.ptys(),
            vec![("xterm-256color".to_owned(), 80, 24, 10, 20)],
            "the family params ride the wire pty request verbatim"
        );
        assert_eq!(observed.shells().len(), 1, "the shell request rides");
        drop(session);
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the pipe-mode shell parity (mirroring the
    ///   server direction's decided row — a pty-less shell is
    ///   legitimate vanilla and served, never declined): the
    ///   `params: None` shell carries **no** pty request.
    #[tokio::test]
    async fn pipe_shell_launches_without_a_pty_request() {
        let (connection, double, observed) = dial_scripted(SshdScript::default()).await;
        let session = connection
            .shell(None)
            .await
            .expect("the pipe shell launches");
        assert!(
            observed.ptys().is_empty(),
            "no pty request rides the pipe-mode shell"
        );
        assert_eq!(observed.shells().len(), 1);
        drop(session);
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the pty control rows riding the **live
    ///   pump** (the POC-3 §D.3 window-change/signal mechanics',
    ///   client arm): the session never `into_stream()`s, so the
    ///   fire-and-forget rows ride the write half mid-session — the
    ///   `CTRL_IN` `Resize` translates to the wire `window_change`
    ///   and `Signal` to the wire `signal` (the double's
    ///   observations are the witnesses).
    #[tokio::test]
    async fn pty_control_rows_ride_the_live_session_pump() {
        let (connection, double, observed) = dial_scripted(SshdScript::default()).await;
        let params = TerminalParamsWire {
            term: Some("xterm-256color".to_owned()),
            cols: 80,
            rows: 24,
            pixel_width: 0,
            pixel_height: 0,
            modes: serde_json::json!({}),
        };
        let mut session = connection
            .shell(Some(params))
            .await
            .expect("the pty shell row launches");

        session
            .input
            .write_all(
                &control_frame(&ControlMessage::Resize {
                    cols: 120,
                    rows: 40,
                    pixel_width: 0,
                    pixel_height: 0,
                })
                .await,
            )
            .await
            .expect("the resize writes");
        session
            .input
            .write_all(
                &control_frame(&ControlMessage::Signal {
                    name: "TERM".to_owned(),
                })
                .await,
            )
            .await
            .expect("the signal writes");

        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
        let witnessed = || {
            observed.windows() == vec![(120, 40, 0, 0)]
                && observed.signals().iter().any(|s| s.contains("TERM"))
        };
        while !witnessed() {
            if tokio::time::Instant::now() >= deadline {
                panic!("the control translations never landed: {observed:?}");
            }
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        assert_eq!(
            observed.windows(),
            vec![(120, 40, 0, 0)],
            "the CTRL_IN Resize translated to the wire window_change"
        );
        assert!(
            observed.signals().iter().any(|s| s.contains("TERM")),
            "the CTRL_IN Signal translated to the wire signal, name verbatim"
        );

        drop(session);
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the family exit-currency's signal mirror
    ///   (mapping.md's derivation contract, the client arm — consumed
    ///   not re-derived): a wire `exit-signal("TERM")` reports the
    ///   family code `−15` in-band (rule 2's value, the table read
    ///   backwards; the derivation never invents).
    #[tokio::test]
    async fn exit_signal_report_mirrors_the_family_code_in_band() {
        let script = SshdScript {
            stdout: b"killed",
            exit: Some(DblExit::Signal("TERM")),
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let mut session = connection
            .exec("start a session")
            .await
            .expect("the exec row launches");
        assert_eq!(observed.execs(), vec![b"start a session".to_vec()]);
        send_stdin_eof(&mut session).await;
        let (stdout, _, exit) = read_to_exit(session).await;
        assert_eq!(stdout, b"killed");
        assert_eq!(
            exit,
            Some(-15),
            "the −signum mirror, table-read, no invention"
        );
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the never-half-capability posture (ADR-003
    ///   decision 3, client mirror) — a **refused pty refuses the
    ///   whole row**: the exec behind it never rides the wire (the
    ///   double's exec observations stay empty).
    #[tokio::test]
    async fn refused_pty_refuses_the_whole_row_never_half_serves() {
        let script = SshdScript {
            request_success: false,
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let params = TerminalParamsWire {
            term: Some("xterm".to_owned()),
            cols: 80,
            rows: 24,
            pixel_width: 0,
            pixel_height: 0,
            modes: serde_json::json!({}),
        };
        let err = connection
            .pty_exec(params, "echo never")
            .await
            .expect_err("the refused pty refuses the row");
        assert!(matches!(err, OpError::RequestRefused), "got {err:?}");
        assert!(
            observed.execs().is_empty() && observed.shells().is_empty(),
            "the requests behind the refused pty never rode the wire (the row's channel closed behind the refusal)"
        );
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the server-rejected **open** row (POC-1 §6,
    ///   client arm) + the connection-stays-usable row (per decline
    ///   class, `client.md` §Correctness facts): the session open
    ///   refused (the scripted count) → `OpenDeclined(Prohibited)`,
    ///   then a later launch on the *same connection* succeeds and
    ///   pumps — the usable-after-decline posture — and everything
    ///   closes clean.
    #[tokio::test]
    async fn server_rejected_open_declines_and_the_connection_stays_usable() {
        let script = SshdScript {
            reject_first_session_opens: 1,
            stdout: b"still-usable",
            exit: Some(DblExit::Status(0)),
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;

        let err = connection
            .exec("echo declined")
            .await
            .expect_err("the server-rejected open declines");
        assert!(
            matches!(err, OpError::OpenDeclined(OpenDeclineReason::Prohibited)),
            "got {err:?}"
        );
        assert!(
            !connection.is_closed(),
            "the decline never poisons the connection"
        );

        let mut session = connection
            .exec("echo still-usable")
            .await
            .expect("the same connection serves the next launch");
        send_stdin_eof(&mut session).await;
        let (stdout, _, exit) = read_to_exit(session).await;
        assert_eq!(stdout, b"still-usable");
        assert_eq!(exit, Some(0));
        assert_eq!(
            observed.execs(),
            vec![b"echo still-usable".to_vec()],
            "the declined open carried no request; the usable launch rode the wire"
        );
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the task's smoke item (the §module docs'
    ///   decided posture; like the wave-3 smokes, skipped-with-reason
    ///   when the tooling is absent): the client exec row driven
    ///   against the alkssh *server direction*'s real session rows —
    ///   both russh stacks over one live wire, the chunk bridge's two
    ///   arms meeting (the client's family plane consuming the
    ///   server's family plane), data and the exit-status
    ///   round-tripping end to end.
    #[tokio::test]
    async fn smoke_client_exec_row_against_the_server_directions_session_rows() {
        if std::process::Command::new("sh")
            .arg("-c")
            .arg("exit 0")
            .output()
            .is_err()
        {
            eprintln!("SKIPPED (with reason): the op-row smoke — sh is unavailable");
            return;
        }

        let provider = Arc::new(common::MemIdentityProvider::new());
        provider.register(
            &fingerprint(&fixture("alpha").public_key()),
            common::identity("op-row-smoke", &["ssh"]),
        );
        let mut serve =
            common::harness::serve_config(provider, &[fixture("bravo")], Limits::default());
        serve.exec_wrapper_argv = Some(vec!["/bin/sh".to_owned(), "-c".to_owned()]);
        {
            let mut backends = std::collections::HashMap::new();
            backends.insert(
                "local".to_owned(),
                Arc::new(alktty::LocalTtyBackend::new()) as Arc<dyn alktty::TtyBackend>,
            );
            serve.session_seam = Some(alkssh::SessionSeam {
                backends: Arc::new(backends),
                backend: "local".to_owned(),
                ownership: None,
            });
        }
        let adapter = Arc::new(
            alkssh::SshAdapter::new(serve)
                .await
                .expect("the smoke adapter constructs"),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("the smoke listener binds");
        let addr = listener.local_addr().expect("the smoke has an address");
        let accept_task = tokio::spawn(async move {
            loop {
                let Ok((stream, _)) = listener.accept().await else {
                    return;
                };
                let adapter = Arc::clone(&adapter);
                tokio::spawn(async move {
                    let _ = adapter.handle(stream).await;
                });
            }
        });

        let policy = Arc::new(MemHostKeyPolicy::new());
        policy.pin("127.0.0.1", fixture("bravo").fingerprint);
        let config = DialConfig::new(
            addr.to_string(),
            "runner",
            alpha_key_source(),
            Arc::clone(&policy) as Arc<dyn alkssh::HostKeyPolicy>,
        );
        let connection = dial(config).await.expect("the smoke dial succeeds");
        let session = connection
            .exec("echo alkssh-op-row-smoke")
            .await
            .expect("the client exec row drives the alkssh server direction");

        let (stdout, stderr, exit) = read_to_exit(session).await;
        assert!(
            String::from_utf8_lossy(&stdout).contains("alkssh-op-row-smoke"),
            "the program's stdout rode both chunk bridges: {stdout:?}"
        );
        assert_eq!(stderr, Vec::<u8>::new(), "pipe mode keeps stderr separate");
        assert_eq!(
            exit,
            Some(0),
            "the exit chunk rode the family plane, both arms"
        );

        accept_task.abort();
        connection.close().await;
    }
}

#[cfg(feature = "tunnel")]
mod tunnel_rows {
    use super::*;
    use alkssh::TunnelChannel;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    /// - *Contract stamp:* `client.md` §op rows (row 2, the typed
    ///   `channel_open_direct_tcpip` open — the alksocks `DialFn`
    ///   composes over this surface) + POC-3 §C.1's recipe reversed —
    ///   the pump is the raw byte duplex, round-tripping, no channel
    ///   queue consumed anywhere between the open and the
    ///   `into_stream()` conversion.
    #[tokio::test]
    async fn direct_tcpip_pump_round_trips_the_raw_duplex() {
        let script = SshdScript {
            direct: DblDirect::Echo,
            ..SshdScript::default()
        };
        let (connection, double, _observed) = dial_scripted(script).await;
        let mut channel: TunnelChannel = connection
            .tunnel_direct("hub.example", 4096)
            .await
            .expect("the typed open succeeds");

        let payload = b"tunnel-payload-round-trip";
        channel.write_all(payload).await.expect("the pump writes");
        channel.flush().await.expect("the pump flushes");
        let mut echo = vec![0u8; payload.len()];
        tokio::time::timeout(Duration::from_secs(10), channel.read_exact(&mut echo))
            .await
            .expect("the echo arrives")
            .expect("the echo reads");
        assert_eq!(echo, payload, "the byte plane round-trips");

        channel.shutdown().await.expect("the EOF rides");
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the server-rejected open's family mirror
    ///   (POC-1 §6 — the typed open resolves `Err` on the reject,
    ///   family-mapped) + the connection-stays-usable row **per
    ///   decline class**: after the `ConnectFailed` class the
    ///   streamlocal twin still opens and pumps through the same
    ///   connection, and the whole thing closes clean.
    #[cfg(unix)]
    #[tokio::test]
    async fn server_rejected_direct_open_leaves_the_connection_usable() {
        let script = SshdScript {
            direct: DblDirect::Reject,
            streamlocal: DblDirect::Echo,
            ..SshdScript::default()
        };
        let (connection, double, _observed) = dial_scripted(script).await;

        let err = connection
            .tunnel_direct("hub.example", 4096)
            .await
            .expect_err("the scripted server refuses the direct open");
        assert!(
            matches!(err, OpError::OpenDeclined(OpenDeclineReason::ConnectFailed)),
            "the wire class mirrors, got {err:?}"
        );
        assert!(
            !connection.is_closed(),
            "the decline never poisons the connection"
        );

        let mut channel = connection
            .tunnel_direct_unix("/run/alkssh-op-row.sock")
            .await
            .expect("the usable-after-decline row: the twin opens on the same connection");
        let payload = b"after-the-decline";
        channel.write_all(payload).await.expect("the pump writes");
        channel.flush().await.expect("the pump flushes");
        let mut echo = vec![0u8; payload.len()];
        tokio::time::timeout(Duration::from_secs(10), channel.read_exact(&mut echo))
            .await
            .expect("the echo arrives")
            .expect("the echo reads");
        assert_eq!(echo, payload);

        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* OQ-11 (the streamlocal twin's runtime-Unix
    ///   gate, client arm — universal compile, clean off-Unix decline
    ///   **before any wire op**): this row compiles only off-Unix;
    ///   the on-Unix builds ride the usable-after-decline row's
    ///   success half.
    #[cfg(not(unix))]
    #[tokio::test]
    async fn streamlocal_twin_declines_off_unix_before_any_wire_op() {
        let (connection, mut double, _observed) = dial_scripted(SshdScript::default()).await;
        let err = connection
            .tunnel_direct_unix("/run/alkssh-off-unix.sock")
            .await
            .expect_err("the OQ-11 gate declines off-Unix");
        assert!(matches!(err, OpError::UnixUnavailable), "got {err:?}");
        no_open_observed(&mut double, "the off-Unix twin's decline").await;
        expect_clean_close(connection, double).await;
    }
}

#[cfg(feature = "git")]
mod git_rows {
    use super::*;
    use alkgit::Service;
    use alkssh::GitPackSession;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    /// The scripted "advertisement" bytes the double's program
    /// writes (a placeholder pack-side payload for the pump row).
    const ADVERT: &[u8] = b"0008version 2\n0000";

    /// - *Contract stamp:* `client.md` §op rows (row 4 — the shared
    ///   exec grammar's command, *reused, not forked*; the wire
    ///   carries `git-upload-pack '/owner/demo'` byte-verbatim) + the
    ///   pack duplex pump + the exit-status consumption.
    #[tokio::test]
    async fn git_row_round_trips_the_pack_duplex_and_consumes_the_exit() {
        let script = SshdScript {
            stdout: ADVERT,
            exit: Some(DblExit::Status(0)),
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let mut pack: GitPackSession = connection
            .git_service(Service::UploadPack, "owner/demo")
            .await
            .expect("the git row opens");
        assert_eq!(
            observed.execs(),
            vec![b"git-upload-pack '/owner/demo'".to_vec()],
            "the shared grammar's command rode the wire byte-verbatim"
        );

        pack.output
            .write_all(b"pack bytes upstream")
            .await
            .expect("the pack pump writes");
        pack.input.shutdown().await.expect("the EOF rides");
        let mut advert = Vec::new();
        tokio::time::timeout(
            Duration::from_secs(10),
            pack.output.read_to_end(&mut advert),
        )
        .await
        .expect("the advertisement arrives")
        .expect("the advertisement reads");
        assert_eq!(advert, ADVERT, "the pack duplex pumped");

        let exit = tokio::time::timeout(Duration::from_secs(10), pack.exit)
            .await
            .expect("the exit report arrives")
            .expect("the pump consumed it");
        assert_eq!(exit, 0, "exit-status(0) consumed into the family code");
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the git row's pre-wire gates riding the
    ///   shared grammar's verdicts (`client.md` §op rows, row 4; the
    ///   git row's server-direction gate mirrored): an id outside the
    ///   ADR-008 grammar declines (`OpError::InvalidRepoId`) **before
    ///   any wire op**.
    #[tokio::test]
    async fn invalid_repo_id_declines_before_any_wire_op() {
        let (connection, mut double, observed) = dial_scripted(SshdScript::default()).await;
        let err = connection
            .git_service(Service::UploadPack, "a/b/c")
            .await
            .expect_err("the validation gate declines");
        assert!(matches!(err, OpError::InvalidRepoId), "got {err:?}");
        assert!(observed.execs().is_empty());
        no_open_observed(&mut double, "the invalid-repo-id decline").await;
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the per-length cap's client mirror (the
    ///   shared grammar's verdict — `Limits::
    ///   max_exec_command_len`): before any wire op.
    #[tokio::test]
    async fn oversized_git_command_declines_before_any_wire_op() {
        let limits = Limits {
            max_exec_command_len: 4,
            ..Limits::default()
        };
        let (connection, mut double, observed) =
            dial_scripted_limits(SshdScript::default(), limits).await;
        let err = connection
            .git_service(Service::ReceivePack, "owner/demo")
            .await
            .expect_err("the length gate declines");
        assert!(matches!(err, OpError::CommandTooLong), "got {err:?}");
        assert!(observed.execs().is_empty());
        no_open_observed(&mut double, "the length gate's decline").await;
        expect_clean_close(connection, double).await;
    }

    /// - *Contract stamp:* the upstream's own refusal is the
    ///   request-refusal class (the drain's `Failure` — POC-3 §D.3's
    ///   git-row shape): declined cleanly, the connection intact.
    #[tokio::test]
    async fn refused_git_exec_surfaces_the_request_refusal() {
        let script = SshdScript {
            request_success: false,
            ..SshdScript::default()
        };
        let (connection, double, observed) = dial_scripted(script).await;
        let err = connection
            .git_service(Service::ReceivePack, "owner/demo")
            .await
            .expect_err("the scripted upstream refuses");
        assert!(matches!(err, OpError::RequestRefused), "got {err:?}");
        assert_eq!(
            observed.execs(),
            vec![b"git-receive-pack '/owner/demo'".to_vec()],
            "the receive-pack command rode the wire (the refusal is its reply)"
        );
        expect_clean_close(connection, double).await;
    }
}