//! The `w3-tty-session` contract rows: the session branch of the
//! spine served for real: pty/shell/pipe-exec through the published
//! alktty seam (`drive_session_pre_negotiated`) over the translation
//! bridge, the locally-built negotiation, the control rows, the exit
//! derivation contract (mapping.md rules 1–6), and the
//! and the negotiation-error decline class; all observed through the
//! loopback harness against the hand-rolled TtyBackend shim
//! (`tests/common/tty_doubles.rs`). The decline-matrix rows *per
//! feature build* stay in `tests/dispatch_core.rs`; this binary adds
//! the session rows' serving surface, the unset-field declines (which
//! need a configured seam to isolate), and the real-tool pty smoke
//! (feature-gated: in a `session`-less build the binary compiles
//! empty, per the crates' compile-time feature isolation).

#![allow(dead_code)]

mod common;

#[cfg(feature = "session")]
mod serving {
    use std::path::PathBuf;
    use std::sync::Arc;

    use alktty::backend::TerminalParams;
    use alktty::wire::MAX_CHUNK_LEN;
    use russh::ChannelOpenFailure;
    use russh::client::Msg;
    use russh::{Channel, ChannelMsg, Error, Sig};

    use super::common::event_capture as ec;
    use super::common::harness::{
        LoopbackSession, authenticate_publickey, drive, next_message, open_session_channel,
        request_exec, request_pty, request_shell, request_subsystem,
        serve_config as harness_serve_config,
    };
    use super::common::tty_doubles as td;
    use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
    use alkssh::{Limits, ServeConfig, SshAdapter};

    /// The observation surface on the wire, the terminal-observation
    /// enum these rows drive (russh-verbatim shapes, drained in order
    /// until the deadline).
    #[derive(Debug)]
    pub(super) enum Observed {
        Data(Vec<u8>),
        Stderr(Vec<u8>),
        ExitStatus(u32),
        ExitSignal { signal_name: Sig, core_dumped: bool },
        Close,
        Failure,
    }

    /// The next observation with a deadline; None past 10 s (nothing
    /// else can arrive; a row hanging past the deadline is its bug).
    pub(super) async fn next_observation(channel: &mut Channel<Msg>) -> Option<Observed> {
        tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                match next_message(channel).await {
                    Some(ChannelMsg::Data { data }) => return Some(Observed::Data(data.to_vec())),
                    Some(ChannelMsg::ExtendedData { data, ext: 1 }) => {
                        return Some(Observed::Stderr(data.to_vec()));
                    }
                    Some(ChannelMsg::ExitStatus { exit_status }) => {
                        return Some(Observed::ExitStatus(exit_status));
                    }
                    Some(ChannelMsg::ExitSignal {
                        signal_name,
                        core_dumped,
                        ..
                    }) => {
                        return Some(Observed::ExitSignal {
                            signal_name,
                            core_dumped,
                        });
                    }
                    Some(ChannelMsg::Close) => return Some(Observed::Close),
                    Some(ChannelMsg::Failure) => return Some(Observed::Failure),
                    Some(_) => continue,
                    None => return None,
                }
            }
        })
        .await
        .ok()
        .flatten()
    }

    /// Drains observations until the terminal outcome (status or
    /// signal), collecting the data rides for the ride rows, and
    /// asserting the close follows after (the status-before-close
    /// order).
    pub(super) async fn collect_terminal(
        channel: &mut Channel<Msg>,
    ) -> (Vec<Vec<u8>>, Vec<Vec<u8>>, Observed) {
        let (mut stdout, mut stderr) = (Vec::new(), Vec::new());
        loop {
            match next_observation(channel).await {
                Some(Observed::Data(data)) => stdout.push(data),
                Some(Observed::Stderr(data)) => stderr.push(data),
                Some(Observed::ExitStatus(code)) => {
                    let terminal = Observed::ExitStatus(code);
                    assert_observed_close_after(channel).await;
                    return (stdout, stderr, terminal);
                }
                Some(Observed::ExitSignal {
                    signal_name,
                    core_dumped,
                }) => {
                    let terminal = Observed::ExitSignal {
                        signal_name,
                        core_dumped,
                    };
                    assert_observed_close_after(channel).await;
                    return (stdout, stderr, terminal);
                }
                other => panic!("expected the terminal outcome, got {other:?}"),
            }
        }
    }

    /// The status-before-close contract's second half: a close must
    /// follow the terminal observation.
    pub(super) async fn assert_observed_close_after(channel: &mut Channel<Msg>) {
        match next_observation(channel).await {
            Some(Observed::Close) => {}
            other => panic!("expected the channel close after the terminal outcome, got {other:?}"),
        }
    }

    /// Waits for one Success reply (want-reply requests), with a
    /// deadline; a Failure or a silent close is the panic, the rows
    /// drive Success paths.
    pub(super) async fn expect_success(channel: &mut Channel<Msg>, context: &str) {
        if tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                match next_message(channel).await {
                    Some(ChannelMsg::Success) => return true,
                    Some(ChannelMsg::Failure) => panic!("{context}: unexpected request failure"),
                    Some(_) => continue,
                    None => panic!("{context}: the channel closed before the Success reply"),
                }
            }
        })
        .await
        .is_err()
        {
            panic!("{context}: the Success reply never arrived");
        }
    }

    /// The shim-allocation poll: the first recorded `TtyParams`, once
    /// the bridge's session actually allocated.
    pub(super) async fn await_allocation(shim: &Arc<td::TtyShim>) -> alktty::backend::TtyParams {
        for _ in 0..200 {
            if let Some(params) = shim.observation().allocations.lock().first() {
                return params.clone();
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        panic!("the shim never allocated (no session reached the seam)");
    }

    /// The shim-resize poll.
    pub(super) async fn await_resize(shim: &Arc<td::TtyShim>) -> (u16, u16) {
        for _ in 0..200 {
            if let Some(resize) = shim.observation().resizes.lock().last() {
                return *resize;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        panic!("no resize reached the shim's control");
    }

    /// The shim-signal poll.
    pub(super) async fn await_signal(shim: &Arc<td::TtyShim>) -> String {
        for _ in 0..200 {
            if let Some(name) = shim.observation().signals.lock().last() {
                return name.clone();
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        panic!("no signal reached the shim's control");
    }

    /// Builds the session rows' serve-config, everything set (the
    /// shim seam, the shell argv, the exec wrapper, session cwd/env),
    /// overridable per row through the mutator. The identity resolves
    /// from the `alpha` fixture into `ssh-rider` (the witness id).
    pub(super) fn session_config(
        provider: Arc<MemIdentityProvider>,
        mutate: impl FnOnce(&mut ServeConfig),
        limits: Limits,
    ) -> ServeConfig {
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], limits);
        serve.shell_argv = Some(vec!["/bin/none".to_owned(), "-login".to_owned()]);
        serve.exec_wrapper_argv = Some(vec!["/bin/wrap".to_owned(), "-x".to_owned()]);
        serve.session_cwd = Some(PathBuf::from("/srv/session"));
        serve
            .session_env
            .insert("TEST_SESSION_ENV".to_owned(), "rides".to_owned());
        mutate(&mut serve);
        serve
    }

    /// Mounts an authenticated loopback session over the given config
    /// (the rows' driver; fixture `alpha` → `ssh-rider`).
    pub(super) async fn session_with(mutate: impl FnOnce(&mut ServeConfig)) -> LoopbackSession {
        session_with_limits(mutate, Limits::default()).await
    }

    /// [`session_with`] over non-default [`Limits`], for the rows that
    /// tune a budget (the channel-cap release row's ceiling of 1).
    pub(super) async fn session_with_limits(
        mutate: impl FnOnce(&mut ServeConfig),
        limits: Limits,
    ) -> LoopbackSession {
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("ssh-rider", &["ssh"]),
        );
        let adapter = SshAdapter::new(session_config(provider, mutate, limits))
            .await
            .expect("the session build's adapter constructs");
        let policy = Arc::new(MemHostKeyPolicy::with_default(true));
        let mut session = drive(adapter, Arc::clone(&policy), Limits::default())
            .await
            .expect("the loopback mounts");
        let result =
            authenticate_publickey(&mut session.client, "ssh-rider", fixture("alpha")).await;
        assert!(
            result.success(),
            "the fixture identity authenticates before every session row"
        );
        session
    }

    /// A shim-backed seam, the rows' default (no ownership provider).
    pub(super) fn shim_seam(shim: &Arc<td::TtyShim>) -> impl FnOnce(&mut ServeConfig) {
        let seam = td::tty_seam(shim);
        move |serve: &mut ServeConfig| serve.session_seam = Some(seam)
    }

    /// Opens a session channel and sends the pty+shell request pair;
    /// replaced per-row below (the rows drive their own sequences).

    #[tokio::test]
    async fn pty_shell_serves_the_negotiated_shell_row_through_the_bridge() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![b"shell-out".to_vec()],
            stderr: vec![],
            exit: Some(0),
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_pty(&channel).await.expect("the pty succeeded");
        expect_success(&mut channel, "the pty request").await;
        request_shell(&channel).await.expect("the shell routed");
        expect_success(&mut channel, "the shell request").await;
        let (stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        assert_eq!(
            stdout.concat(),
            b"shell-out".to_vec(),
            "the scripted stdout chunk rode SSH data"
        );
        let Observed::ExitStatus(0) = terminal else {
            panic!("the exit chunk derived exit-status(0), got {terminal:?}")
        };

        let params = await_allocation(&shim).await;
        assert_eq!(
            params.cmd,
            vec!["/bin/none".to_owned(), "-login".to_owned()],
            "the shell argv rode the negotiation verbatim (OQ-08)"
        );
        let TerminalParams {
            term,
            cols,
            rows,
            pixel_width,
            pixel_height,
            ..
        } = params.terminal.expect("the pty params reached the seam");
        assert_eq!(term.as_deref(), Some("xterm"));
        assert_eq!((cols, rows), (80, 24), "the pty dims rode");
        assert_eq!((pixel_width, pixel_height), (0, 0));
        assert_eq!(
            params.cwd.as_deref(),
            Some(std::path::Path::new("/srv/session")),
            "session_cwd rides the negotiation, not the wire"
        );
        assert_eq!(
            params.env.get("TEST_SESSION_ENV").map(String::as_str),
            Some("rides"),
            "session_env rides the negotiation, not the wire"
        );
        assert!(
            params.backend_params.is_empty(),
            "no backend selector fields exist on this door"
        );

        session.unmount().await;
    }

    #[tokio::test]
    async fn pipe_exec_serves_the_wrapper_appended_command_in_pipe_mode() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![b"outs".to_vec()],
            stderr: vec![b"errd".to_vec()],
            exit: Some(0),
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_exec(&channel, "uptime")
            .await
            .expect("the exec routed");
        expect_success(&mut channel, "the routed exec").await;
        let (stdout, stderr, terminal) = collect_terminal(&mut channel).await;
        assert_eq!(stdout.concat(), b"outs".to_vec(), "stdout rode SSH data");
        assert_eq!(
            stderr.concat(),
            b"errd".to_vec(),
            "stderr rode SSH extended_data(1)"
        );
        let Observed::ExitStatus(0) = terminal else {
            panic!("the exit chunk derived exit-status(0), got {terminal:?}")
        };

        let params = await_allocation(&shim).await;
        assert_eq!(
            params.cmd,
            vec!["/bin/wrap".to_owned(), "-x".to_owned(), "uptime".to_owned()],
            "the wire command appended to the wrapper argv, never naively split"
        );
        assert!(
            params.terminal.is_none(),
            "no pty preceded: the exec row serves pipe mode"
        );

        session.unmount().await;
    }

    #[tokio::test]
    async fn pty_then_exec_serves_pty_mode_with_the_stored_params() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: Some(0),
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_pty(&channel).await.expect("the pty succeeded");
        expect_success(&mut channel, "the pty request").await;
        request_exec(&channel, "echo done").await.expect("routed");
        expect_success(&mut channel, "the routed exec").await;
        let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        let Observed::ExitStatus(0) = terminal else {
            panic!("exit-status(0), got {terminal:?}")
        };

        let params = await_allocation(&shim).await;
        assert_eq!(
            params.cmd,
            vec![
                "/bin/wrap".to_owned(),
                "-x".to_owned(),
                "echo done".to_owned()
            ]
        );
        let TerminalParams { cols, rows, .. } =
            params.terminal.expect("pty-then-exec serves PTY mode");
        assert_eq!(
            (cols, rows),
            (80, 24),
            "`ssh -t host \"cmd\"` serves PTY-mode"
        );

        session.unmount().await;
    }

    #[tokio::test]
    async fn shell_without_a_pty_serves_pipe_mode() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: Some(0),
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_shell(&channel).await.expect("the shell routed");
        expect_success(&mut channel, "the shell request").await;
        let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        let Observed::ExitStatus(0) = terminal else {
            panic!("exit-status(0), got {terminal:?}")
        };

        let params = await_allocation(&shim).await;
        assert!(
            params.terminal.is_none(),
            "the decided row: a shell without a preceding pty serves pipe-mode"
        );

        session.unmount().await;
    }

    // ------------------------------------------------------------------
    // The channel-not-held guard (`w4-channel-not-held-rows`): a second
    // request on an already-served channel declines request-level; the
    // store-and-take handoff holds a channel exactly once, the wave-3
    // gate's booked finding 2.
    // ------------------------------------------------------------------

    /// - *Contract stamp:* mapping.md §Request-level rows (the exec
    ///   row's store-and-take handoff, the `channel-not-held` class:
    ///   a second exec on an already-served pipe channel answers
    ///   `channel_failure`, inline, the first serve rides on) + the
    ///   event vocabulary (`src/observe.rs`: class `request`, reason
    ///   `channel-not-held`, route `pipe`; the exit-`None` script
    ///   keeps the serve live and the wire quiescent, so the row is
    ///   race-free by construction).
    #[tokio::test]
    async fn second_exec_on_a_served_pipe_channel_declines_channel_not_held() {
        ec::install();
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_exec(&channel, "uptime")
            .await
            .expect("the first exec routed");
        expect_success(&mut channel, "the served pipe exec").await;

        request_exec(&channel, "uptime")
            .await
            .expect("the second exec rides the wire");
        match next_message(&mut channel).await {
            Some(ChannelMsg::Failure) => {}
            other => {
                panic!("a second exec on a served channel answers channel_failure, got {other:?}")
            }
        }

        let event = ec::await_event(&[
            ("row", "exec_request"),
            ("decline_class", "request"),
            ("reason", "channel-not-held"),
            ("exec_route", "pipe"),
        ])
        .await;
        ec::assert_request_decline(&event, true);

        request_subsystem(&channel, "sftp")
            .await
            .expect("the alive channel carries another request");
        match next_message(&mut channel).await {
            Some(ChannelMsg::Failure) => {}
            other => panic!("the inline decline leaves the channel usable, got {other:?}"),
        }

        session.unmount().await;
    }

    /// - *Contract stamp:* mapping.md §Request-level rows (the
    ///   `shell_request` row's store-and-take handoff: the
    ///   `channel-not-held` class: a second shell request on an
    ///   already-served shell channel answers `channel_failure`,
    ///   inline, the first serve rides on) + the event vocabulary
    ///   (`src/observe.rs`: class `request`, reason
    ///   `channel-not-held`; the lifecycle row carries no
    ///   `exec_route`; absent-field shape pinned on the wire).
    #[tokio::test]
    async fn second_shell_request_on_a_served_shell_channel_declines_channel_not_held() {
        ec::install();
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_shell(&channel)
            .await
            .expect("the first shell request routed");
        expect_success(&mut channel, "the served shell row").await;

        request_shell(&channel)
            .await
            .expect("the second shell request rides the wire");
        match next_message(&mut channel).await {
            Some(ChannelMsg::Failure) => {}
            other => panic!(
                "a second shell request on a served channel answers channel_failure, \
                 got {other:?}"
            ),
        }

        let event = ec::await_event(&[
            ("row", "shell_request"),
            ("decline_class", "request"),
            ("reason", "channel-not-held"),
        ])
        .await;
        ec::assert_request_decline(&event, false);

        request_subsystem(&channel, "sftp")
            .await
            .expect("the alive channel carries another request");
        match next_message(&mut channel).await {
            Some(ChannelMsg::Failure) => {}
            other => panic!("the inline decline leaves the channel usable, got {other:?}"),
        }

        session.unmount().await;
    }

    // ------------------------------------------------------------------
    // The channel_close-fired per-channel cancel path on the wire
    // (`w4-tty-coverage`, wave-3 gate booked item 3): the cap release +
    // the per-channel state teardown, wire-witnessed; the disjoint
    // release sites the gate verified per row (the wire rows here
    // witness the established-channel-close site, `channel_cap.rs`
    // release site 1).
    // ------------------------------------------------------------------

    /// - *Contract stamp:* mapping.md §Bounded resources /
    ///   `channel_cap.rs` release site 1 + `handler.rs`
    ///   §the control rows' per-channel surface (a client close on a
    ///   served row fires the `channel_close` hook: the bridge surface
    ///   fires the per-channel cancel; the serving task aborts the
    ///   alktty session, alktty's kill-on-drop contract fires, nothing
    ///   fabricated on the wire, and the cap releases, disjoint from
    ///   the bridge-failure and arrival sites).
    #[tokio::test]
    async fn a_client_close_on_a_served_row_fires_the_per_channel_cancel_and_releases_the_cap() {
        ec::install();
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with_limits(
            shim_seam(&shim),
            Limits {
                max_concurrent_channels: 1,
                ..Limits::default()
            },
        )
        .await;

        let mut served = open_session_channel(&session.client)
            .await
            .expect("the serving channel opens (the cap admits)");
        request_exec(&served, "uptime")
            .await
            .expect("the exec routed");
        expect_success(&mut served, "the served pipe exec").await;

        let refused = open_session_channel(&session.client)
            .await
            .expect_err("beyond the cap of 1: refused at the open");
        assert!(
            matches!(
                refused,
                Error::ChannelOpenFailure(ChannelOpenFailure::AdministrativelyProhibited)
            ),
            "the cap-exceeded refusal is the AdministrativelyProhibited open decline, \
             got {refused:?}"
        );

        served.close().await.expect("the client close rides");

        for _ in 0..200 {
            if !shim.observation().kills.lock().is_empty() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(
            !shim.observation().kills.lock().is_empty(),
            "the close-fired per-channel cancel aborted the alktty session; \
             alktty's kill-on-drop contract fired (witnessed before unmount, \
             so this is the channel_close path, not the teardown)"
        );
        match next_observation(&mut served).await {
            Some(Observed::Close) | None => {}
            other => panic!("the cancel fabricates no exit report, got {other:?}"),
        }
        let _ = ec::await_event(&[
            ("row", "exec_request"),
            ("message", "serving path ended"),
            ("exec_route", "pipe"),
        ])
        .await;

        let mut recovered = open_session_channel(&session.client)
            .await
            .expect("the cap recovered after the close, since the channel_close hook released");
        request_exec(&recovered, "uptime")
            .await
            .expect("the recovered channel routes");
        expect_success(&mut recovered, "the recovered channel serves").await;

        session.unmount().await;
    }

    // ------------------------------------------------------------------
    // The Closed/Break no-exit-chunk close shape on the wire
    // (`w4-tty-coverage`): a bridge read that ends without an exit
    // chunk reports nothing; the derivation contract's absent-report
    // posture pinned, not just the happy paths. (The frame-level
    // Closed/Break classification rows live in `src/tty_map.rs`'s test
    // mod; the Break classification is unreachable through alktty's
    // own validated `ChunkWriter`, a defensive arm.)
    // ------------------------------------------------------------------

    /// - *Contract stamp:* mapping.md §Exit chunk (the contract's
    ///   close-shape class: the bridge read ends **without** an exit
    ///   chunk → no report, plain close, the lifecycle's ended event:
    ///   its absence is neither 0 nor 255). The shape is constructed
    ///   through the real path: the shim's stdout carries a payload
    ///   beyond alktty's own `MAX_CHUNK_LEN` wire bound; alktty's
    ///   `ChunkWriter` rejects it locally, its drainer shuts the write
    ///   half down, and the bridge's next read is EOF (`Closed`) with
    ///   the scripted exit never forwarded. Nothing fabricated.
    #[tokio::test]
    async fn a_bridge_end_without_an_exit_chunk_reports_nothing_and_closes_plain() {
        ec::install();
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![b"bridge-ok".to_vec(), vec![0u8; MAX_CHUNK_LEN as usize + 1]],
            stderr: vec![],
            exit: Some(0),
        });
        let session = session_with(shim_seam(&shim)).await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "uptime")
            .await
            .expect("the exec routed");
        expect_success(&mut channel, "the routed exec").await;

        assert!(
            !shim.observation().allocations.lock().is_empty(),
            "the session truly served (the no-report close shape, not a decline)"
        );
        let mut saw_data = false;
        let mut saw_close = false;
        loop {
            match next_observation(&mut channel).await {
                Some(Observed::Data(data)) if data == b"bridge-ok".to_vec() => {
                    saw_data = true;
                }
                Some(Observed::Close) => {
                    saw_close = true;
                    break;
                }
                Some(Observed::ExitStatus(_)) | Some(Observed::ExitSignal { .. }) => panic!(
                    "the no-exit-chunk close shape fabricates no exit report, \
                     the scripted exit never rode"
                ),
                None => break,
                Some(Observed::Data(other)) => {
                    panic!("unexpected data after the chunk plane died: {other:?}")
                }
                Some(_) => continue,
            }
        }
        assert!(
            saw_data,
            "the in-bounds chunk rode SSH data before the chunk plane died"
        );
        assert!(saw_close, "the Closed shape ends with a plain close");
        let _ = ec::await_event(&[
            ("row", "exec_request"),
            ("message", "serving path ended"),
            ("exec_route", "pipe"),
        ])
        .await;

        session.unmount().await;
    }

    #[tokio::test]
    async fn stdin_bytes_and_the_eof_sentinel_reach_the_backend() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with(shim_seam(&shim)).await;

        let channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_shell(&channel).await.expect("the shell routed");
        let _ = await_allocation(&shim).await;

        channel
            .data_bytes(&b"typed data"[..])
            .await
            .expect("the stdin ride");
        let data_reached = |shim: &Arc<td::TtyShim>| {
            let log = shim.observation().stdin.lock().clone();
            log == b"typed data".to_vec()
        };
        for _ in 0..200 {
            if data_reached(&shim) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(
            data_reached(&shim),
            "SSH data → Chunk::stdin reached the backend verbatim"
        );

        channel.eof().await.expect("the channel EOF rides the wire");
        for _ in 0..200 {
            if !shim.observation().stdin_eof.lock().is_empty() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(
            !shim.observation().stdin_eof.lock().is_empty(),
            "the SSH EOF translated to the zero-length stdin-chunk sentinel: \
             the backend's stdin shutdown was observed"
        );

        // The session never resolves its exit (script exit = None);
        // the disconnect is the connection-level cancel: the serving
        // task aborts the alktty session (the kill-on-drop witness
        // recorded, nothing fabricated on the wire.
        session.unmount().await;
        for _ in 0..100 {
            if !shim.observation().kills.lock().is_empty() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(
            !shim.observation().kills.lock().is_empty(),
            "the cancellation aborted the never-resolving session; \
             alktty's kill-on-drop contract fired"
        );
    }

    #[tokio::test]
    async fn window_change_reaches_the_bridge_and_preserve_resize_updates_the_stored_params() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with(shim_seam(&shim)).await;

        // Pre-serve resize: after the pty is stored, before the shell
        // request; the stored params resize (the final size wins).
        let channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_pty(&channel).await.expect("the pty succeeded");
        channel
            .window_change(120, 40, 0, 0)
            .await
            .expect("the window-change rides (fire-and-forget)");
        request_shell(&channel).await.expect("the shell routed");
        let params = await_allocation(&shim).await;
        let TerminalParams { cols, rows, .. } =
            params.terminal.expect("the stored pty served the session");
        assert_eq!((cols, rows), (120, 40), "the pre-serve resize won");

        // Serving resize: the bridge forwards through the bounded
        // control surface.
        channel
            .window_change(200, 50, 0, 0)
            .await
            .expect("the window-change rides");
        let (resized_cols, resized_rows) = await_resize(&shim).await;
        assert_eq!(
            (resized_cols, resized_rows),
            (200, 50),
            "`window_change_request` → `ControlMessage::Resize` CTRL_IN"
        );

        session.unmount().await;
    }

    #[tokio::test]
    async fn signal_request_forwards_the_name_verbatim_to_the_backend() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let session = session_with(shim_seam(&shim)).await;

        let channel = open_session_channel(&session.client)
            .await
            .expect("the session channel opens");
        request_pty(&channel).await.expect("the pty succeeded");
        request_shell(&channel).await.expect("the shell routed");
        let _ = await_allocation(&shim).await;

        channel
            .signal(Sig::TERM)
            .await
            .expect("the signal rides (fire-and-forget)");
        let name = await_signal(&shim).await;
        assert_eq!(name, "TERM", "accepted verbatim, no alkssh-side filter");

        session.unmount().await;
    }

    // ------------------------------------------------------------------
    // The exit derivation contract (mapping.md rules 1–6), on the wire.
    // ------------------------------------------------------------------

    /// - *Contract stamp:* mapping.md §Exit chunk rules 1, 2, 6 (code
    ///   ≥ 0 → `exit-status(N)`; negative `−N`, N ≥ 2 →
    ///   `exit-signal(name)` via the table; POC-3 §D.2's `−15` →
    ///   `TERM` live evidence) + rules 3 (the `−1` sentinel → 255,
    ///   never a signal claim); each rule's wire row follows. `core_dumped`
    ///   is never asserted (the bool stays false).
    #[tokio::test]
    async fn exit_chunks_derive_by_the_contract_rules() {
        for exit_code in [0, 1, -15, -2, -33, -1] {
            let shim = Arc::new(td::TtyShim::new());
            shim.set_script(td::ShimScript {
                stdout: vec![],
                stderr: vec![b"e".to_vec()],
                exit: Some(exit_code),
            });
            let session = session_with(shim_seam(&shim)).await;
            let mut channel = open_session_channel(&session.client)
                .await
                .expect("the channel opens");
            request_exec(&channel, "noop").await.expect("routed");
            expect_success(&mut channel, "the routed exec").await;
            let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
            match (exit_code, &terminal) {
                (code, Observed::ExitStatus(status)) if code >= 0 && *status == code as u32 => {}
                (
                    -15,
                    Observed::ExitSignal {
                        signal_name,
                        core_dumped,
                    },
                ) => {
                    assert!(matches!(signal_name, Sig::TERM), "rule 2: −15 → TERM");
                    assert!(
                        !core_dumped,
                        "core_dumped is never asserted; the bool stays false"
                    );
                }
                (
                    -2,
                    Observed::ExitSignal {
                        signal_name,
                        core_dumped,
                    },
                ) => {
                    assert!(matches!(signal_name, Sig::INT), "rule 2: −2 → INT");
                    assert!(!core_dumped);
                }
                (-33, Observed::ExitStatus(255)) => {}
                (-1, Observed::ExitStatus(255)) => {}
                _ => panic!(
                    "exit chunk {exit_code}: expected the derived contract shape, got {terminal:?}"
                ),
            }
            session.unmount().await;
        }
    }

    // ------------------------------------------------------------------
    // The negotiation-error decline class (the bridge's serving-path
    // decline: the session never started → exit-status(255) + close).
    // ------------------------------------------------------------------

    /// - *Contract stamp:* mapping.md §the seams + the task's
    ///   negotiate-then-serve decision (the alktty error frame →
    ///   `negotiation-failed`, `exit-status(255)`, close: the one
    ///   could-not-start claim this door makes: nothing ran).
    #[tokio::test]
    async fn unknown_backend_answers_255_then_close() {
        let session = session_with(|serve: &mut ServeConfig| {
            serve.session_seam = Some(td::misconfigured_seam());
        })
        .await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "uptime").await.expect("routed");
        let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        let Observed::ExitStatus(255) = terminal else {
            panic!("could-not-start reports exit-status(255), got {terminal:?}")
        };
        session.unmount().await;
    }

    #[tokio::test]
    async fn ownership_denial_answers_255_then_close_and_witnesses_the_identity() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: None,
        });
        let ownership = Arc::new(td::MemOwnership::new(false));
        let witness = Arc::clone(&ownership);
        let session = session_with(|serve: &mut ServeConfig| {
            serve.session_seam = Some(td::tty_seam_with_ownership(&shim, Arc::clone(&ownership)));
        })
        .await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "uptime").await.expect("routed");
        let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        let Observed::ExitStatus(255) = terminal else {
            panic!("the denied ownership is could-not-start (255), got {terminal:?}")
        };
        let ids = witness.ids.lock().clone();
        assert_eq!(
            ids,
            vec!["ssh-rider".to_owned()],
            "the resolved identity rode to drive_session_pre_negotiated; \
             the parked wave-2 identity-landing item asserted live"
        );
        session.unmount().await;
    }

    #[tokio::test]
    async fn the_identity_rides_to_the_seam_on_a_serving_row() {
        let shim = Arc::new(td::TtyShim::new());
        shim.set_script(td::ShimScript {
            stdout: vec![],
            stderr: vec![],
            exit: Some(0),
        });
        let ownership = Arc::new(td::MemOwnership::new(true));
        let witness = Arc::clone(&ownership);
        let session = session_with(move |serve: &mut ServeConfig| {
            serve.session_seam = Some(td::tty_seam_with_ownership(&shim, ownership));
        })
        .await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "noop").await.expect("routed");
        let (_stdout, _stderr, terminal) = collect_terminal(&mut channel).await;
        let Observed::ExitStatus(0) = terminal else {
            panic!("exit-status(0), got {terminal:?}")
        };
        let ids = witness.ids.lock().clone();
        assert!(
            ids.iter().any(|id| id == "ssh-rider"),
            "the identity reached the seam's ownership check on the serving path"
        );
        session.unmount().await;
    }
}

#[cfg(feature = "session")]
mod unset_declines {
    use std::sync::Arc;

    use super::common::harness::{
        LoopbackSession, authenticate_publickey, drive, next_message, open_session_channel,
        request_exec, request_pty, request_shell, serve_config as harness_serve_config,
    };
    use super::common::tty_doubles as td;
    use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
    use alkssh::{Limits, ServeConfig, SshAdapter};
    use russh::client::Msg;
    use russh::{Channel, ChannelMsg};

    /// An authenticated loopback over a mutated config, the
    /// unset-field rows' driver.
    async fn session_with(mutate: impl FnOnce(&mut ServeConfig)) -> LoopbackSession {
        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("unset-rider", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default());
        serve.exec_wrapper_argv = Some(vec!["/bin/wrap".to_owned()]);
        serve.shell_argv = Some(vec!["/bin/none".to_owned()]);
        mutate(&mut serve);
        let adapter = SshAdapter::new(serve)
            .await
            .expect("the adapter constructs");
        let policy = Arc::new(MemHostKeyPolicy::with_default(true));
        let mut session = drive(adapter, Arc::clone(&policy), Limits::default())
            .await
            .expect("the loopback mounts");
        let result =
            authenticate_publickey(&mut session.client, "unset-rider", fixture("alpha")).await;
        assert!(result.success());
        session
    }

    async fn expect_failure_then_close(channel: &mut Channel<Msg>, context: &str) {
        match next_message(channel).await {
            Some(ChannelMsg::Failure) => {}
            other => panic!("{context}: expected failure, got {other:?}"),
        }
        loop {
            match next_message(channel).await {
                Some(ChannelMsg::Close) => return,
                None => panic!("{context}: expected the close after the failure"),
                Some(_) => continue,
            }
        }
    }

    /// - *Contract stamp:* OQ-08 (`shell_argv` unset → `shell_request`
    ///   declines, never a fallback) + the recorded wire shape (the
    ///   serving-path decline: failure then close, from a spawned
    ///   task).
    #[tokio::test]
    async fn shell_argv_unset_declines_the_shell_row() {
        let shim = Arc::new(td::TtyShim::new());
        let shim_for_seam = Arc::clone(&shim);
        let session = session_with(move |serve: &mut ServeConfig| {
            serve.shell_argv = None;
            serve.session_seam = Some(td::tty_seam(&shim_for_seam));
        })
        .await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_shell(&channel).await.expect("rides the wire");
        expect_failure_then_close(&mut channel, "the unset shell row").await;
        assert!(
            shim.observation().allocations.lock().is_empty(),
            "a declined row never allocates (decline-first, alloc-never)"
        );
        session.unmount().await;
    }

    /// - *Contract stamp:* OQ-08 (`exec_wrapper_argv` unset → the
    ///   pipe row declines) + the recorded wire shape.
    #[tokio::test]
    async fn exec_wrapper_unset_declines_the_pipe_row() {
        let shim = Arc::new(td::TtyShim::new());
        let shim_for_seam = Arc::clone(&shim);
        let session = session_with(move |serve: &mut ServeConfig| {
            serve.exec_wrapper_argv = None;
            serve.session_seam = Some(td::tty_seam(&shim_for_seam));
        })
        .await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "uptime").await.expect("rides");
        expect_failure_then_close(&mut channel, "the unset wrapper row").await;
        assert!(
            shim.observation().allocations.lock().is_empty(),
            "a declined row never allocates"
        );
        session.unmount().await;
    }

    /// - *Contract stamp:* the decline matrix's config-driven note
    ///   (`session_seam` unset → the session rows decline even though
    ///   the `session` feature is on) + the recorded wire shape; the
    ///   pty row declines too (a doomed allocation is declined first,
    ///   alloc never).
    #[tokio::test]
    async fn seam_unset_declines_every_session_row() {
        let session = session_with(|_serve: &mut ServeConfig| {}).await;
        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_pty(&channel).await.expect("rides the wire");
        expect_failure_then_close(&mut channel, "the unset-seam pty row").await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_shell(&channel).await.expect("rides the wire");
        expect_failure_then_close(&mut channel, "the unset-seam shell row").await;

        let mut channel = open_session_channel(&session.client)
            .await
            .expect("the channel opens");
        request_exec(&channel, "uptime").await.expect("rides");
        expect_failure_then_close(&mut channel, "the unset-seam pipe row").await;
        session.unmount().await;
    }
}

#[cfg(all(unix, feature = "session"))]
mod real_tool {
    //! The real-tool smoke (`w3-tty-session`): one OpenSSH pty
    //! session (`ssh -tt`, a real PTY via a dev-dep alktty `local`
    //! composition, the posture decided: the session rows' smoke
    //! composes the REAL backend, not the shim, because a plain shim
    //! row cannot observe what vanilla OpenSSH actually sends; POC-3
    //! stage D proved the same composition) through a TCP-mounted
    //! adapter. Posture (recorded): requires `ssh` and `sh` in the
    //! environment; when absent it is **skipped with a printed
    //! reason** (skipped-with-reason at most, never silently green).
    //! The smoke owns the pty lifecycle rows only (data rides, clean
    //! exit → exit-status before close; the `stty size`-grade
    //! fidelity matrix is wave 5's).

    use std::collections::HashMap;
    use std::process::Command;
    use std::sync::Arc;

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

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

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

    #[tokio::test]
    async fn real_openssh_pty_session_smoke() {
        if Command::new("ssh").arg("-V").output().is_err() {
            eprintln!("SKIPPED (with reason): real-tool pty smoke, ssh is unavailable");
            return;
        }
        if Command::new("sh").arg("-c").arg("exit 0").output().is_err() {
            eprintln!("SKIPPED (with reason): real-tool pty smoke, sh is unavailable");
            return;
        }

        let work = std::env::temp_dir().join(format!("alkssh-tty-smoke-{}", std::process::id()));
        std::fs::create_dir_all(&work).expect("the smoke workdir creates");
        let key_path = work.join("id_ed25519");
        std::fs::write(&key_path, fixture("alpha").pem).expect("the fixture key writes");
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600))
            .expect("the key file perms set");

        let provider = Arc::new(MemIdentityProvider::new());
        provider.register(
            fixture("alpha").fingerprint,
            identity("smoke-rider", &["ssh"]),
        );
        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Limits::default());
        serve.shell_argv = Some(vec!["/bin/sh".to_owned(), "-i".to_owned()]);
        serve.session_seam = Some(local_seam());
        let adapter = Arc::new(
            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 port = listener.local_addr().expect("local addr").port();
        let accept_task = tokio::spawn(async move {
            loop {
                let Ok((stream, _)) = listener.accept().await else {
                    return;
                };
                let adapter = Arc::clone(&adapter);
                tokio::spawn(async move {
                    let _ = adapter.handle(stream).await;
                });
            }
        });

        let outcome = tokio::time::timeout(std::time::Duration::from_secs(30), async {
            let mut child = tokio::process::Command::new("ssh")
                .args(["-tt", "-p", &port.to_string(), "-i"])
                .arg(&key_path)
                .args([
                    "-o",
                    "BatchMode=yes",
                    "-o",
                    "IdentitiesOnly=yes",
                    "-o",
                    "StrictHostKeyChecking=no",
                    "-o",
                    "UserKnownHostsFile=/dev/null",
                    "-o",
                    "ConnectTimeout=10",
                ])
                .arg("smoke-rider@127.0.0.1")
                .stdin(std::process::Stdio::piped())
                .stdout(std::process::Stdio::piped())
                .spawn()
                .expect("ssh spawns");
            {
                use tokio::io::AsyncWriteExt;
                let mut stdin = child.stdin.take().expect("the child stdin is piped");
                stdin
                    .write_all(b"echo alkssh-tty-smoke; exit\n")
                    .await
                    .expect("the smoke stdin writes");
                stdin.flush().await.expect("the smoke stdin flushes");
            }
            let output = child.wait_with_output().await.expect("ssh completes");
            (
                output.status.success(),
                String::from_utf8_lossy(&output.stdout).to_string(),
            )
        })
        .await;

        accept_task.abort();
        let _ = std::fs::remove_dir_all(&work);

        let (success, stdout) = match outcome {
            Ok((success, stdout)) => (success, stdout),
            Err(_) => panic!("the real ssh pty smoke exceeded its 30 s deadline"),
        };
        assert!(
            success,
            "vanilla OpenSSH's pty+shell session through the door exits 0 \
             (the exit chunk derived exit-status(0) before close); stdout: {stdout:?}"
        );
    }
}