nyaterm/src-tauri/src/core/ssh/io.rs

use super::client::{
    SshDiagnosticContext, SshDiagnosticStage, SshHandle, SshPostLoginConfig, SshStartupCommand,
};
use crate::config::SftpCwdFollowMode;
use crate::core::capture::OutputCaptureProcessor;
use crate::core::monitoring::stats::RemoteStatsSampler;
use crate::core::ssh::osc::{self, OscStripper, ShellKind};
use crate::core::terminal_session::local::split_startup_passthrough;
use crate::core::terminal_session::{
    TerminalOutputDecoder, encode_terminal_input, prepare_terminal_write_input,
};
use crate::core::zmodem::{
    ZmodemAction, ZmodemDetectResult, ZmodemDetector, ZmodemDirection, ZmodemDownloadOoDrain,
    ZmodemEvent, ZmodemTransfer, ZmodemUploadDrain, start_zmodem_transfer,
};
use crate::core::{
    InputOrigin, RecordingManager, SessionCommand, SessionCommandReceiver, SessionCommandSender,
    SessionCwdReplacement, SessionManager, SessionOutputCoalescer, SharedCwd, replace_cwd_state,
    update_cwd_if_changed,
};
use crate::error::{AppError, AppResult};
use russh::{ChannelMsg, client};
use std::{pin::Pin, sync::Arc, time::Instant};
use tauri::{AppHandle, Emitter, Manager};
use tokio::time::{Duration, Sleep, timeout};

const INJECT_TIMEOUT_SECS: u64 = 8;
const INITIAL_INJECT_DELAY_MS: u64 = 500;
const SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES: usize = 64 * 1024;
const SUPPRESSION_DIAGNOSTIC_INITIAL_MS: u64 = 1_000;
const SUPPRESSION_DIAGNOSTIC_INTERVAL_MS: u64 = 2_000;

#[derive(Debug)]
enum ShellDetectionResult {
    Detected { shell: ShellKind },
    NoSupportedShell,
    TimedOut,
    ChannelOpenFailed,
    ExecFailed,
}

/// Tries to detect the remote shell via an exec channel with a timeout.
///
/// Produces the same runtime outcome as the previous `Option<ShellKind>`
/// contract: only `Detected` enables integration, all other outcomes skip it.
async fn detect_shell_type<H: client::Handler>(
    handle: &mut client::Handle<H>,
    session_id: &str,
    timeout_ms: u64,
) -> ShellDetectionResult {
    tracing::info!(
        session_id = %session_id,
        timeout_ms,
        "SSH shell detection starting"
    );

    let started = Instant::now();
    let fut = async {
        let mut ch = match handle.channel_open_session().await {
            Ok(channel) => channel,
            Err(error) => {
                return Err(("channel_open", error.to_string(), 0usize));
            }
        };
        tracing::info!(
            session_id = %session_id,
            "SSH shell detection channel opened"
        );

        if let Err(error) = ch
            .exec(
                true,
                r#"printf '%s\n' "$SHELL"; ps -p $$ -o comm= 2>/dev/null || true"#,
            )
            .await
        {
            return Err(("exec", error.to_string(), 0usize));
        }
        tracing::info!(
            session_id = %session_id,
            "SSH shell detection exec request sent"
        );

        let mut output = String::new();
        let mut output_bytes = 0usize;
        while let Some(msg) = ch.wait().await {
            if let ChannelMsg::Data { ref data } = msg {
                output_bytes += data.len();
                output.push_str(&String::from_utf8_lossy(data));
            }
        }

        Ok((ShellKind::from_name(output.trim()), output_bytes))
    };

    match timeout(Duration::from_millis(timeout_ms), fut).await {
        Ok(Ok((shell, output_bytes))) if shell != ShellKind::Unknown => {
            tracing::info!(
                session_id = %session_id,
                shell = ?shell,
                elapsed_ms = started.elapsed().as_millis() as u64,
                output_bytes,
                "SSH shell detection completed"
            );
            ShellDetectionResult::Detected { shell }
        }
        Ok(Ok((_shell, output_bytes))) => {
            tracing::info!(
                session_id = %session_id,
                elapsed_ms = started.elapsed().as_millis() as u64,
                output_bytes,
                "SSH shell detection returned no supported shell"
            );
            ShellDetectionResult::NoSupportedShell
        }
        Ok(Err(("channel_open", error, _output_bytes))) => {
            tracing::warn!(
                session_id = %session_id,
                elapsed_ms = started.elapsed().as_millis() as u64,
                %error,
                "SSH shell detection channel open failed"
            );
            ShellDetectionResult::ChannelOpenFailed
        }
        Ok(Err(("exec", error, _output_bytes))) => {
            tracing::warn!(
                session_id = %session_id,
                elapsed_ms = started.elapsed().as_millis() as u64,
                %error,
                "SSH shell detection exec failed"
            );
            ShellDetectionResult::ExecFailed
        }
        Ok(Err((_stage, error, _output_bytes))) => {
            tracing::warn!(
                session_id = %session_id,
                elapsed_ms = started.elapsed().as_millis() as u64,
                %error,
                "SSH shell detection failed"
            );
            ShellDetectionResult::NoSupportedShell
        }
        Err(_) => {
            tracing::warn!(
                session_id = %session_id,
                timeout_ms,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "SSH shell detection timed out"
            );
            ShellDetectionResult::TimedOut
        }
    }
}

async fn exec_remote_command<H: client::Handler>(
    handle: &mut client::Handle<H>,
    command: &str,
    timeout_ms: u64,
) -> AppResult<String> {
    let fut = async {
        let mut ch = handle.channel_open_session().await.map_err(|error| {
            AppError::Channel(format!("Failed to open exec channel: {}", error))
        })?;
        ch.exec(true, command).await.map_err(|error| {
            AppError::Channel(format!("Failed to execute remote command: {}", error))
        })?;

        let mut output = String::new();
        let mut exit_status = None;
        while let Some(msg) = ch.wait().await {
            match msg {
                ChannelMsg::Data { ref data } | ChannelMsg::ExtendedData { ref data, .. } => {
                    output.push_str(&String::from_utf8_lossy(data));
                }
                ChannelMsg::ExitStatus {
                    exit_status: status,
                } => {
                    exit_status = Some(status);
                }
                _ => {}
            }
        }

        match exit_status.unwrap_or(0) {
            0 => Ok(output),
            status => Err(AppError::Channel(format!(
                "Remote command exited with status {status}: {output}"
            ))),
        }
    };

    timeout(Duration::from_millis(timeout_ms), fut)
        .await
        .map_err(|_| AppError::Channel("Remote command timed out".to_string()))?
}

fn sh_single_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\\''"))
}

fn remote_install_command(shell: ShellKind) -> Option<String> {
    let script = osc::persistent_script(shell)?;
    let block = osc::rc_managed_block(shell)?;
    let script_path = osc::persistent_script_path(shell)?;
    let rc_path = osc::rc_file_path(shell)?;

    Some(format!(
        r#"set -eu
script_path={script_path}
rc_path={rc_path}
mkdir -p "$HOME/.config/nyaterm"
case "$rc_path" in */*) mkdir -p "${{rc_path%/*}}" ;; esac
script_tmp="${{script_path}}.tmp.$$"
cat > "$script_tmp" <<'NYATERM_SCRIPT_EOF'
{script}
NYATERM_SCRIPT_EOF
if [ ! -f "$script_path" ] || ! cmp -s "$script_tmp" "$script_path"; then
  mv "$script_tmp" "$script_path"
else
  rm -f "$script_tmp"
fi
block_tmp="${{script_path}}.block.$$"
cat > "$block_tmp" <<'NYATERM_BLOCK_EOF'
{block}
NYATERM_BLOCK_EOF
rc_tmp="${{rc_path}}.tmp.$$"
start={start}
end={end}
if [ -f "$rc_path" ] && grep -F "$start" "$rc_path" >/dev/null 2>&1 && grep -F "$end" "$rc_path" >/dev/null 2>&1; then
  NYATERM_BLOCK_FILE="$block_tmp" awk -v start="$start" -v end="$end" '
    $0 == start {{
      if (!done) {{
        while ((getline line < ENVIRON["NYATERM_BLOCK_FILE"]) > 0) print line
        close(ENVIRON["NYATERM_BLOCK_FILE"])
        done=1
      }}
      skip=1
      next
    }}
    $0 == end {{ skip=0; next }}
    !skip {{ print }}
    END {{
      if (!done) {{
        if (NR > 0) print ""
        while ((getline line < ENVIRON["NYATERM_BLOCK_FILE"]) > 0) print line
      }}
    }}
  ' "$rc_path" > "$rc_tmp"
else
  if [ -f "$rc_path" ]; then
    cat "$rc_path" > "$rc_tmp"
    if [ -s "$rc_tmp" ]; then printf '\n' >> "$rc_tmp"; fi
  else
    : > "$rc_tmp"
  fi
  cat "$block_tmp" >> "$rc_tmp"
fi
if [ ! -f "$rc_path" ] || ! cmp -s "$rc_tmp" "$rc_path"; then
  if [ -f "$rc_path" ] && [ ! -f "$rc_path.nyaterm.bak" ]; then
    cp "$rc_path" "$rc_path.nyaterm.bak" 2>/dev/null || true
  fi
  mv "$rc_tmp" "$rc_path"
else
  rm -f "$rc_tmp"
fi
rm -f "$block_tmp"
"#,
        script_path = script_path,
        rc_path = rc_path,
        script = script,
        block = block,
        start = sh_single_quote(osc::MANAGED_BLOCK_START),
        end = sh_single_quote(osc::MANAGED_BLOCK_END),
    ))
}

async fn install_remote_shell_integration<H: client::Handler>(
    handle: &mut client::Handle<H>,
    shell: ShellKind,
) -> AppResult<()> {
    let Some(command) = remote_install_command(shell) else {
        return Err(AppError::Config(format!(
            "No persistent shell integration available for {shell:?}"
        )));
    };
    exec_remote_command(handle, &command, 3000)
        .await
        .map(|_| ())
}

const CHANNEL_REQUEST_REPLY_TIMEOUT_MS: u64 = 10_000;

async fn wait_channel_request_reply(
    channel: &mut russh::Channel<client::Msg>,
    request_name: &str,
) -> AppResult<()> {
    let reply = timeout(
        Duration::from_millis(CHANNEL_REQUEST_REPLY_TIMEOUT_MS),
        async {
            loop {
                match channel.wait().await {
                    Some(ChannelMsg::Success) => return Ok(()),
                    Some(ChannelMsg::Failure) => {
                        return Err(AppError::Channel(format!(
                            "{request_name} request rejected by server"
                        )));
                    }
                    Some(ChannelMsg::Close | ChannelMsg::Eof) | None => {
                        return Err(AppError::Channel(format!(
                            "SSH channel closed before {request_name} request completed"
                        )));
                    }
                    Some(_) => {}
                }
            }
        },
    )
    .await;

    match reply {
        Ok(result) => result,
        Err(_) => Err(AppError::Channel(format!(
            "{request_name} request timed out"
        ))),
    }
}

async fn close_failed_interactive_channel(
    channel: &russh::Channel<client::Msg>,
    session_id: &str,
    stage: &str,
) {
    if let Err(error) = channel.close().await {
        tracing::debug!(
            session_id = %session_id,
            stage,
            %error,
            "Failed to close rejected SSH interactive channel"
        );
    }
}

/// Opens a PTY shell channel and detects the remote shell type.
///
/// Returns `(channel, Option<injection_script>, ready_marker)`.
/// The injection script is **not** sent here — the IO loop defers it until
/// the shell has produced its initial output (banner / MOTD) so that the
/// welcome text is not swallowed.
pub(super) async fn open_shell_channel<H: client::Handler>(
    handle: &mut client::Handle<H>,
    session_id: &str,
    x11_fake_cookie_hex: Option<&str>,
    agent_forwarding: bool,
    terminal_type: &str,
    sftp_enabled: bool,
    network_device_profile: bool,
    cwd_follow_mode: SftpCwdFollowMode,
    shell_detection_timeout_ms: u64,
    diagnostics: Option<SshDiagnosticContext>,
) -> AppResult<(
    russh::Channel<client::Msg>,
    Option<String>,
    String,
    Option<ShellKind>,
    Option<String>,
)> {
    if let Some(diagnostics) = diagnostics.as_ref() {
        diagnostics.set_stage(SshDiagnosticStage::OpeningInteractiveChannel);
    }
    tracing::info!(
        session_id = %session_id,
        "SSH interactive channel opening"
    );
    let mut channel = match handle.channel_open_session().await {
        Ok(channel) => {
            tracing::info!(
                session_id = %session_id,
                "SSH interactive channel opened"
            );
            channel
        }
        Err(error) => {
            tracing::warn!(
                session_id = %session_id,
                %error,
                "SSH interactive channel open failed"
            );
            return Err(AppError::Channel(format!(
                "Failed to open channel: {}",
                error
            )));
        }
    };

    let mut local_notice = None;
    if agent_forwarding {
        if let Err(error) = channel.agent_forward(false).await {
            tracing::warn!(
                session_id = %session_id,
                %error,
                "Could not enable SSH Agent forwarding"
            );
        }
    }
    if let Some(fake_cookie_hex) = x11_fake_cookie_hex {
        match channel
            .request_x11(true, false, "MIT-MAGIC-COOKIE-1", fake_cookie_hex, 0)
            .await
        {
            Ok(()) => {
                if let Err(error) = wait_channel_request_reply(&mut channel, "X11").await {
                    tracing::warn!(
                        session_id = %session_id,
                        %error,
                        "Could not enable X11 forwarding"
                    );
                    local_notice = Some(super::x11_forwarding::enable_failed_message());
                }
            }
            Err(error) => {
                tracing::warn!(
                    session_id = %session_id,
                    %error,
                    "Could not enable X11 forwarding"
                );
                local_notice = Some(super::x11_forwarding::enable_failed_message());
            }
        }
    }

    if let Some(diagnostics) = diagnostics.as_ref() {
        diagnostics.set_stage(SshDiagnosticStage::RequestingPty);
    }
    tracing::info!(
        session_id = %session_id,
        terminal = %terminal_type,
        cols = 80,
        rows = 24,
        "SSH PTY request sending"
    );
    if let Err(error) = channel
        .request_pty(true, terminal_type, 80, 24, 0, 0, &[])
        .await
    {
        tracing::warn!(
            session_id = %session_id,
            %error,
            "SSH PTY request failed"
        );
        close_failed_interactive_channel(&channel, session_id, "pty-send").await;
        return Err(AppError::Channel(format!("PTY request failed: {}", error)));
    }
    if let Err(error) = wait_channel_request_reply(&mut channel, "PTY").await {
        tracing::warn!(
            session_id = %session_id,
            %error,
            "SSH PTY request rejected"
        );
        close_failed_interactive_channel(&channel, session_id, "pty-reply").await;
        return Err(error);
    }
    tracing::info!(
        session_id = %session_id,
        "SSH PTY request completed"
    );

    if let Some(diagnostics) = diagnostics.as_ref() {
        diagnostics.set_stage(SshDiagnosticStage::RequestingShell);
    }
    tracing::info!(
        session_id = %session_id,
        "SSH shell request sending"
    );
    if let Err(error) = channel.request_shell(true).await {
        tracing::warn!(
            session_id = %session_id,
            %error,
            "SSH shell request failed"
        );
        close_failed_interactive_channel(&channel, session_id, "shell-send").await;
        return Err(AppError::Channel(format!(
            "Shell request failed: {}",
            error
        )));
    }
    if let Err(error) = wait_channel_request_reply(&mut channel, "Shell").await {
        tracing::warn!(
            session_id = %session_id,
            %error,
            "SSH shell request rejected"
        );
        close_failed_interactive_channel(&channel, session_id, "shell-reply").await;
        return Err(error);
    }
    tracing::info!(
        session_id = %session_id,
        "SSH shell request completed"
    );

    if let Some(diagnostics) = diagnostics.as_ref() {
        diagnostics.set_stage(SshDiagnosticStage::PreparingIntegration);
    }
    let ready_marker = osc::build_ready_marker(session_id);

    let mut detected_shell = None;
    let injection_script = match cwd_follow_mode {
        SftpCwdFollowMode::Off => {
            let reason = if network_device_profile {
                "network_device_profile"
            } else if sftp_enabled {
                "cwd_follow_disabled"
            } else {
                "sftp_disabled"
            };
            tracing::info!(
                session_id = %session_id,
                mode = ?cwd_follow_mode,
                detected_shell = ?detected_shell,
                has_injection_script = false,
                has_ready_marker = false,
                reason,
                "SSH shell integration skipped"
            );
            None
        }
        SftpCwdFollowMode::ShellIntegration | SftpCwdFollowMode::RcFile => {
            if let Some(diagnostics) = diagnostics.as_ref() {
                diagnostics.set_stage(SshDiagnosticStage::DetectingShell);
            }
            let detection_result =
                detect_shell_type(handle, session_id, shell_detection_timeout_ms).await;
            if let Some(diagnostics) = diagnostics.as_ref() {
                diagnostics.set_stage(SshDiagnosticStage::PreparingIntegration);
            }
            match detection_result {
                ShellDetectionResult::Detected { shell: shell_kind } => {
                    detected_shell = Some(shell_kind);
                    let script = if cwd_follow_mode == SftpCwdFollowMode::RcFile {
                        match install_remote_shell_integration(handle, shell_kind).await {
                            Ok(()) => {
                                tracing::debug!(
                                    session_id = %session_id,
                                    shell = ?shell_kind,
                                    "Remote shell integration files are installed"
                                );
                                osc::activation_script(shell_kind, &ready_marker)
                            }
                            Err(error) => {
                                tracing::warn!(
                                    session_id = %session_id,
                                    shell = ?shell_kind,
                                    %error,
                                    "Failed to install remote shell integration files; falling back to session injection"
                                );
                                osc::injection_script(shell_kind, &ready_marker)
                            }
                        }
                    } else {
                        osc::injection_script(shell_kind, &ready_marker)
                    };
                    if script.is_some() {
                        tracing::info!(
                            session_id = %session_id,
                            mode = ?cwd_follow_mode,
                            detected_shell = ?detected_shell,
                            has_injection_script = true,
                            has_ready_marker = true,
                            "SSH shell integration prepared"
                        );
                    } else {
                        tracing::info!(
                            session_id = %session_id,
                            mode = ?cwd_follow_mode,
                            detected_shell = ?detected_shell,
                            has_injection_script = false,
                            has_ready_marker = false,
                            reason = "no_script_available",
                            "SSH shell integration skipped"
                        );
                    }
                    script
                }
                ShellDetectionResult::TimedOut => {
                    tracing::info!(
                        session_id = %session_id,
                        mode = ?cwd_follow_mode,
                        detected_shell = ?detected_shell,
                        has_injection_script = false,
                        has_ready_marker = false,
                        reason = "shell_detection_timeout",
                        "SSH shell integration skipped"
                    );
                    None
                }
                ShellDetectionResult::NoSupportedShell => {
                    tracing::info!(
                        session_id = %session_id,
                        mode = ?cwd_follow_mode,
                        detected_shell = ?detected_shell,
                        has_injection_script = false,
                        has_ready_marker = false,
                        reason = "shell_unknown",
                        "SSH shell integration skipped"
                    );
                    None
                }
                ShellDetectionResult::ChannelOpenFailed | ShellDetectionResult::ExecFailed => {
                    tracing::info!(
                        session_id = %session_id,
                        mode = ?cwd_follow_mode,
                        detected_shell = ?detected_shell,
                        has_injection_script = false,
                        has_ready_marker = false,
                        reason = "shell_unknown",
                        "SSH shell integration skipped"
                    );
                    None
                }
            }
        }
    };

    Ok((
        channel,
        injection_script,
        ready_marker,
        detected_shell,
        local_notice,
    ))
}

/// Injection phase state machine.
///
/// ```text
/// ┌─────────────┐  first data   ┌─────────────┐  ready marker  ┌────────┐
/// │ WaitInitial  │──────────────▶│ Suppressing  │──────────────▶│ Normal │
/// └─────────────┘  (inject sent) └─────────────┘               └────────┘
///                                       │ timeout
///                                       └──────────────────────▶│ Normal │
/// ```
///
/// When no injection script is provided we start directly in `Normal`.
#[derive(Debug, PartialEq)]
enum IoPhase {
    /// Passing through all output; waiting for the first data chunk so we
    /// can display the banner / MOTD before injecting.
    WaitInitial,
    /// Injection script sent; discarding visible output (echo) until the
    /// ready marker appears.
    Suppressing,
    /// Normal operation — strip our OSC sequences, forward everything else.
    Normal,
}

impl std::fmt::Display for IoPhase {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::WaitInitial => f.write_str("WaitInitial"),
            Self::Suppressing => f.write_str("Suppressing"),
            Self::Normal => f.write_str("Normal"),
        }
    }
}

#[derive(Debug, PartialEq, Eq)]
enum InjectionEvent {
    None,
    Inject,
    Ready { visible_after_ready: String },
    Failed { visible_after_ready: String },
    RuntimeFailed { visible: String },
}

#[derive(Debug, PartialEq, Eq)]
enum InjectionTimeoutEvent {
    None,
    FallbackToNormal,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum InjectionTimeoutSource {
    Deadline,
    WallClock,
}

impl std::fmt::Display for InjectionTimeoutSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Deadline => f.write_str("deadline"),
            Self::WallClock => f.write_str("wall_clock"),
        }
    }
}

#[derive(Debug, Default)]
struct SuppressionDiagnostics {
    rx_bytes_total: u64,
    rx_chunks: u64,
    first_rx_after_ms: Option<u64>,
    last_rx_after_ms: Option<u64>,
    pre_ready_write_bytes: u64,
    pre_ready_write_chunks: u64,
    last_diagnostic_at: Option<Instant>,
}

impl SuppressionDiagnostics {
    fn reset(&mut self) {
        *self = Self::default();
    }

    fn record_rx(&mut self, bytes: usize, injection_sent_at: &Instant, now: Instant) {
        let elapsed_ms = elapsed_ms_at(injection_sent_at, now);
        self.rx_bytes_total = self
            .rx_bytes_total
            .saturating_add(u64::try_from(bytes).unwrap_or(u64::MAX));
        self.rx_chunks = self.rx_chunks.saturating_add(1);
        self.first_rx_after_ms.get_or_insert(elapsed_ms);
        self.last_rx_after_ms = Some(elapsed_ms);
    }

    fn record_pre_ready_write(&mut self, bytes: usize) {
        self.pre_ready_write_bytes = self
            .pre_ready_write_bytes
            .saturating_add(u64::try_from(bytes).unwrap_or(u64::MAX));
        self.pre_ready_write_chunks = self.pre_ready_write_chunks.saturating_add(1);
    }

    fn should_log_suppression_diagnostic(
        &mut self,
        injection_sent_at: &Instant,
        now: Instant,
    ) -> bool {
        if now.saturating_duration_since(*injection_sent_at)
            < Duration::from_millis(SUPPRESSION_DIAGNOSTIC_INITIAL_MS)
        {
            return false;
        }

        if self.last_diagnostic_at.is_some_and(|last| {
            now.saturating_duration_since(last)
                < Duration::from_millis(SUPPRESSION_DIAGNOSTIC_INTERVAL_MS)
        }) {
            return false;
        }

        self.last_diagnostic_at = Some(now);
        true
    }
}

struct PendingStartupCommand {
    input: Vec<u8>,
    delay_ms: u64,
}

/// Build renderer-supplied startup input without allowing terminal controls.
pub(super) fn build_startup_command_input(command: &str) -> Option<Vec<u8>> {
    if command.trim().is_empty() || contains_terminal_control(command) {
        return None;
    }

    let mut input = command.as_bytes().to_vec();
    if !input.ends_with(b"\r") {
        input.push(b'\r');
    }
    Some(input)
}

fn build_post_login_command_input(command: &str) -> Option<Vec<u8>> {
    if command.trim().is_empty() {
        return None;
    }

    let normalized = command.replace("\r\n", "\r").replace('\n', "\r");
    let mut input = normalized.into_bytes();
    if !input.ends_with(b"\r") {
        input.push(b'\r');
    }
    Some(input)
}

fn contains_terminal_control(value: &str) -> bool {
    value
        .chars()
        .any(|ch| matches!(ch, '\u{0000}'..='\u{001F}' | '\u{007F}'..='\u{009F}'))
}

fn arm_post_login_timer(
    phase: &IoPhase,
    pending_post_login: &Option<PendingStartupCommand>,
    post_login_deadline: &mut Option<Pin<Box<Sleep>>>,
) {
    if *phase != IoPhase::Normal {
        return;
    }

    if post_login_deadline.is_none() {
        if let Some(pending) = pending_post_login.as_ref() {
            *post_login_deadline = Some(Box::pin(tokio::time::sleep(Duration::from_millis(
                pending.delay_ms,
            ))));
        }
    }
}

fn should_send_initial_injection(phase: &IoPhase, has_pending_script: bool) -> bool {
    *phase == IoPhase::WaitInitial && has_pending_script
}

fn cancel_pending_injection_for_input(
    phase: &mut IoPhase,
    pending_script: &mut Option<String>,
    origin: InputOrigin,
) -> bool {
    if *phase != IoPhase::WaitInitial
        || pending_script.is_none()
        || origin == InputOrigin::TerminalResponse
    {
        return false;
    }

    pending_script.take();
    *phase = IoPhase::Normal;
    true
}

async fn handle_input_before_initial_injection(
    phase: &mut IoPhase,
    pending_script: &mut Option<String>,
    origin: InputOrigin,
    manager: &SessionManager,
    session_id: &str,
    pending_post_login: &Option<PendingStartupCommand>,
    post_login_deadline: &mut Option<Pin<Box<Sleep>>>,
) {
    if !cancel_pending_injection_for_input(phase, pending_script, origin) {
        return;
    }

    tracing::info!(
        session_id = %session_id,
        ?origin,
        phase_transition = "WaitInitial -> Normal",
        "SSH shell integration injection cancelled by terminal input"
    );
    manager
        .set_dynamic_title_integration_active(session_id, false)
        .await;
    arm_post_login_timer(phase, pending_post_login, post_login_deadline);
}

fn on_initial_injection_sent(phase: &mut IoPhase) {
    if *phase == IoPhase::WaitInitial {
        *phase = IoPhase::Suppressing;
    }
}

fn handle_injection_result(phase: &mut IoPhase, result: &osc::OscResult) -> InjectionEvent {
    match phase {
        IoPhase::WaitInitial => InjectionEvent::Inject,
        IoPhase::Suppressing if result.ready_failed => {
            *phase = IoPhase::Normal;
            InjectionEvent::Failed {
                visible_after_ready: result.visible_after_ready.clone(),
            }
        }
        IoPhase::Suppressing if result.ready => {
            *phase = IoPhase::Normal;
            InjectionEvent::Ready {
                visible_after_ready: result.visible_after_ready.clone(),
            }
        }
        IoPhase::Suppressing => InjectionEvent::None,
        IoPhase::Normal if result.ready_failed => InjectionEvent::RuntimeFailed {
            visible: result.visible.clone(),
        },
        IoPhase::Normal => InjectionEvent::None,
    }
}

fn handle_injection_timeout(phase: &mut IoPhase) -> InjectionTimeoutEvent {
    match phase {
        IoPhase::Suppressing => {
            *phase = IoPhase::Normal;
            InjectionTimeoutEvent::FallbackToNormal
        }
        IoPhase::WaitInitial | IoPhase::Normal => InjectionTimeoutEvent::None,
    }
}

fn elapsed_ms_at(started_at: &Instant, now: Instant) -> u64 {
    u64::try_from(now.saturating_duration_since(*started_at).as_millis()).unwrap_or(u64::MAX)
}

fn injection_has_timed_out_at(
    phase: &IoPhase,
    injection_sent_at: Option<&Instant>,
    now: Instant,
) -> bool {
    *phase == IoPhase::Suppressing
        && injection_sent_at.is_some_and(|started| {
            now.saturating_duration_since(*started) >= Duration::from_secs(INJECT_TIMEOUT_SECS)
        })
}

fn injection_has_timed_out(phase: &IoPhase, injection_sent_at: Option<&Instant>) -> bool {
    injection_has_timed_out_at(phase, injection_sent_at, Instant::now())
}

fn append_suppressed_visible_and_take_passthrough(buffer: &mut String, visible: &str) -> String {
    buffer.push_str(visible);
    let (mut suppressed, passthrough) = split_startup_passthrough(buffer);
    while suppressed.len() > SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES {
        let Some(first_char) = suppressed.chars().next() else {
            break;
        };
        suppressed.drain(..first_char.len_utf8());
    }
    *buffer = suppressed;
    passthrough
}

fn discard_suppressed_output(buffer: &mut String, flushed_osc_buffer: String) -> usize {
    let discarded_len = buffer.len() + flushed_osc_buffer.len();
    buffer.clear();
    discarded_len
}

#[allow(clippy::too_many_arguments)]
async fn send_shell_integration_injection(
    channel: &mut russh::Channel<client::Msg>,
    pending_script: &mut Option<String>,
    inject_deadline: &mut Pin<&mut Sleep>,
    phase: &mut IoPhase,
    session_id: &str,
    shell_kind: Option<ShellKind>,
    injection_sent_at: &mut Option<Instant>,
    suppression_diagnostics: &mut SuppressionDiagnostics,
) {
    let Some(script) = pending_script.take() else {
        return;
    };
    let script_bytes = script.len();
    let script_lines = script.lines().count();
    let max_line_bytes = script.lines().map(str::len).max().unwrap_or_default();

    tracing::info!(
        session_id = %session_id,
        shell = ?shell_kind,
        phase = %phase,
        script_bytes,
        script_lines,
        max_line_bytes,
        "SSH shell integration injection sending"
    );

    match channel.data(script.as_bytes()).await {
        Ok(()) => {
            let sent_at = Instant::now();
            *injection_sent_at = Some(sent_at);
            suppression_diagnostics.reset();
            on_initial_injection_sent(phase);
            inject_deadline
                .as_mut()
                .reset(tokio::time::Instant::now() + Duration::from_secs(INJECT_TIMEOUT_SECS));
            tracing::info!(
                session_id = %session_id,
                shell = ?shell_kind,
                script_bytes,
                script_lines,
                max_line_bytes,
                phase_transition = "WaitInitial -> Suppressing",
                "SSH shell integration injection sent"
            );
        }
        Err(error) => {
            let previous_phase = phase.to_string();
            *phase = IoPhase::Normal;
            *injection_sent_at = None;
            let phase_transition = format!("{previous_phase} -> {phase}");
            tracing::warn!(
                session_id = %session_id,
                shell = ?shell_kind,
                error = %error,
                phase_transition = %phase_transition,
                "SSH shell integration injection send failed; continuing in passive mode"
            );
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn fallback_shell_integration_timeout(
    phase: &mut IoPhase,
    stripper: &mut OscStripper,
    suppressed_visible_fallback: &mut String,
    session_id: &str,
    shell_kind: Option<ShellKind>,
    injection_sent_at: Option<&Instant>,
    suppression_diagnostics: &SuppressionDiagnostics,
    timeout_source: InjectionTimeoutSource,
    pending_post_login: &Option<PendingStartupCommand>,
    post_login_deadline: &mut Option<Pin<Box<Sleep>>>,
) -> bool {
    if *phase != IoPhase::Suppressing || injection_sent_at.is_none() {
        return false;
    }

    let previous_phase = phase.to_string();
    let timeout_event = handle_injection_timeout(phase);
    debug_assert_eq!(timeout_event, InjectionTimeoutEvent::FallbackToNormal);
    let flushed = stripper.flush();
    let osc_buffer_bytes = flushed.len();
    let suppressed_visible_bytes = suppressed_visible_fallback.len();
    let discarded_visible_bytes = discard_suppressed_output(suppressed_visible_fallback, flushed);
    let elapsed_ms = injection_sent_at
        .map(|started| elapsed_ms_at(started, Instant::now()))
        .unwrap_or(0);
    let phase_transition = format!("{previous_phase} -> {phase}");
    tracing::warn!(
        session_id = %session_id,
        shell = ?shell_kind,
        elapsed_ms,
        injection_timeout_secs = INJECT_TIMEOUT_SECS,
        suppression_rx_bytes = suppression_diagnostics.rx_bytes_total,
        suppression_rx_chunks = suppression_diagnostics.rx_chunks,
        first_suppression_rx_after_ms = ?suppression_diagnostics.first_rx_after_ms,
        last_suppression_rx_after_ms = ?suppression_diagnostics.last_rx_after_ms,
        suppressed_visible_bytes,
        osc_buffer_bytes,
        discarded_visible_bytes,
        pre_ready_write_bytes = suppression_diagnostics.pre_ready_write_bytes,
        pre_ready_write_chunks = suppression_diagnostics.pre_ready_write_chunks,
        timeout_source = %timeout_source,
        phase_transition = %phase_transition,
        "SSH shell integration injection timed out"
    );
    arm_post_login_timer(phase, pending_post_login, post_login_deadline);
    true
}

#[allow(clippy::too_many_arguments)]
async fn handle_osc_result(
    app: &AppHandle,
    output: &Arc<SessionOutputCoalescer>,
    cwd_event: &str,
    cwd: &SharedCwd,
    recording_mgr: &Option<Arc<RecordingManager>>,
    session_id: &str,
    manager: &Arc<SessionManager>,
    channel: &mut russh::Channel<client::Msg>,
    pending_script: &mut Option<String>,
    inject_deadline: &mut Pin<&mut Sleep>,
    phase: &mut IoPhase,
    result: &osc::OscResult,
    suppressed_visible_fallback: &mut String,
    shell_kind: Option<ShellKind>,
    injection_sent_at: &mut Option<Instant>,
    suppression_diagnostics: &mut SuppressionDiagnostics,
) {
    let was_suppressing = *phase == IoPhase::Suppressing;
    let injection_event = handle_injection_result(phase, result);
    let suppressed_passthrough = if was_suppressing {
        let before_ready_len = result
            .visible
            .len()
            .saturating_sub(result.visible_after_ready.len());
        append_suppressed_visible_and_take_passthrough(
            suppressed_visible_fallback,
            &result.visible[..before_ready_len],
        )
    } else {
        String::new()
    };
    match injection_event {
        InjectionEvent::Inject => {
            emit_output(
                app,
                output,
                cwd_event,
                cwd,
                recording_mgr,
                session_id,
                manager,
                result,
            )
            .await;
            send_shell_integration_injection(
                channel,
                pending_script,
                inject_deadline,
                phase,
                session_id,
                shell_kind,
                injection_sent_at,
                suppression_diagnostics,
            )
            .await;
        }
        InjectionEvent::Ready {
            visible_after_ready,
        } => {
            let suppressed_visible_bytes = result
                .visible
                .len()
                .saturating_sub(visible_after_ready.len())
                + suppressed_visible_fallback.len();
            suppressed_visible_fallback.clear();
            tracing::info!(
                session_id = %session_id,
                shell = ?shell_kind,
                elapsed_ms = injection_sent_at
                    .as_ref()
                    .map(|started| started.elapsed().as_millis() as u64)
                    .unwrap_or(0),
                suppressed_visible_bytes,
                ready_after_visible_bytes = visible_after_ready.len(),
                phase_transition = "Suppressing -> Normal",
                "SSH shell integration ready marker received"
            );
            // Exact OSC 0/2 and terminal queries retain their original order
            // while source-command echo remains suppressed.
            let mut release_output = suppressed_passthrough;
            release_output.push_str(&visible_after_ready);
            emit_metadata(app, cwd_event, cwd, manager, session_id, result).await;
            emit_visible_text(output, recording_mgr, session_id, &release_output);
        }
        InjectionEvent::Failed {
            visible_after_ready,
        } => {
            let suppressed_visible_bytes = result
                .visible
                .len()
                .saturating_sub(visible_after_ready.len())
                + suppressed_visible_fallback.len();
            suppressed_visible_fallback.clear();
            let mut release_output = suppressed_passthrough;
            release_output.push_str(&visible_after_ready);
            tracing::warn!(
                session_id = %session_id,
                shell = ?shell_kind,
                elapsed_ms = injection_sent_at
                    .as_ref()
                    .map(|started| started.elapsed().as_millis() as u64)
                    .unwrap_or(0),
                suppressed_visible_bytes,
                phase_transition = "Suppressing -> Normal",
                "SSH shell integration reported installation failure; continuing in passive mode"
            );
            emit_metadata(app, cwd_event, cwd, manager, session_id, result).await;
            emit_visible_text(output, recording_mgr, session_id, &release_output);
        }
        InjectionEvent::RuntimeFailed { visible } => {
            tracing::warn!(
                session_id = %session_id,
                shell = ?shell_kind,
                "SSH shell integration failed after startup; clearing stale cwd"
            );
            emit_metadata(app, cwd_event, cwd, manager, session_id, result).await;
            let changes = replace_cwd_state(
                cwd,
                SessionCwdReplacement {
                    legacy_path: None,
                    operational_path: None,
                    presentation: None,
                },
            )
            .await;
            if changes.operational_changed {
                let _ = app.emit(cwd_event, "");
            }
            emit_visible_text(output, recording_mgr, session_id, &visible);
        }
        InjectionEvent::None if *phase == IoPhase::Normal => {
            emit_output(
                app,
                output,
                cwd_event,
                cwd,
                recording_mgr,
                session_id,
                manager,
                result,
            )
            .await;
        }
        InjectionEvent::None => {
            // Transport terminal queries and bounded OSC 0/2 in their exact
            // source order while all remaining startup text stays hidden.
            emit_visible_text(output, recording_mgr, session_id, &suppressed_passthrough);
            emit_metadata(app, cwd_event, cwd, manager, session_id, result).await;
        }
    }
}

pub(super) async fn ssh_io_loop(
    app: AppHandle,
    session_id: String,
    manager: Arc<SessionManager>,
    mut channel: russh::Channel<client::Msg>,
    _handle: SshHandle,
    mut cmd_rx: SessionCommandReceiver,
    output_control_tx: SessionCommandSender,
    cwd: SharedCwd,
    connection_id: Option<String>,
    injection_script: Option<String>,
    ready_marker: String,
    shell_kind: Option<ShellKind>,
    post_login: Option<SshPostLoginConfig>,
    startup_command: Option<SshStartupCommand>,
    backspace_mode: String,
    initial_notice: Option<String>,
    encoding: String,
    diagnostics: Option<SshDiagnosticContext>,
) {
    let backspace_as_bs = backspace_mode == "ctrl_h";
    let output_event = format!("terminal-output-{}", session_id);
    let cwd_event = format!("cwd-changed-{}", session_id);
    let closed_event = format!("session-closed-{}", session_id);

    let recording_mgr: Option<Arc<RecordingManager>> = app
        .try_state::<Arc<RecordingManager>>()
        .map(|state| state.inner().clone());
    let output =
        SessionOutputCoalescer::for_app(app.clone(), output_event.clone(), output_control_tx);
    let mut output_decoder = TerminalOutputDecoder::new(&encoding);
    if let Some(notice) = initial_notice {
        output.push_owned(notice);
    }
    let mut stripper = OscStripper::new(&ready_marker);

    let mut capture_processor = OutputCaptureProcessor::new();

    let has_injection_script = injection_script.is_some();
    let mut phase = if has_injection_script {
        IoPhase::WaitInitial
    } else {
        IoPhase::Normal
    };
    if let Some(diagnostics) = diagnostics.as_ref() {
        diagnostics.set_stage(SshDiagnosticStage::IoRunning);
    }
    tracing::info!(
        session_id = %session_id,
        initial_phase = %phase,
        has_injection_script,
        shell_kind = ?shell_kind,
        "SSH IO loop started"
    );
    let mut suppressed_visible_fallback = String::new();
    let mut pending_script = injection_script;
    let mut initial_remote_data_logged = false;
    let mut injection_sent_at: Option<Instant> = None;
    let mut suppression_diagnostics = SuppressionDiagnostics::default();
    let mut pending_post_login = post_login.and_then(|config| {
        build_post_login_command_input(&config.command).map(|input| PendingStartupCommand {
            input,
            delay_ms: config.delay_ms,
        })
    });
    let mut pending_startup_command = startup_command.and_then(|config| {
        build_startup_command_input(&config.command).map(|input| PendingStartupCommand {
            input,
            delay_ms: config.delay_ms,
        })
    });
    if pending_post_login.is_none() {
        pending_post_login = pending_startup_command.take();
    }
    let mut post_login_deadline: Option<Pin<Box<Sleep>>> = None;
    arm_post_login_timer(&phase, &pending_post_login, &mut post_login_deadline);
    let mut remote_exit_status: Option<u32> = None;
    let mut remote_exit_signal: Option<String> = None;
    let mut output_paused = false;

    let mut zmodem_detector = ZmodemDetector::new();
    let mut zmodem_transfer: Option<ZmodemTransfer> = None;
    let mut zmodem_upload_drain = ZmodemUploadDrain::new();
    let mut zmodem_download_oo_drain = ZmodemDownloadOoDrain::new();
    let zmodem_event_name = format!("zmodem-event-{session_id}");

    let initial_inject_deadline =
        tokio::time::sleep(std::time::Duration::from_millis(INITIAL_INJECT_DELAY_MS));
    tokio::pin!(initial_inject_deadline);

    let inject_deadline = tokio::time::sleep(std::time::Duration::from_secs(INJECT_TIMEOUT_SECS));
    tokio::pin!(inject_deadline);

    let close_reason = loop {
        if injection_has_timed_out(&phase, injection_sent_at.as_ref()) {
            fallback_shell_integration_timeout(
                &mut phase,
                &mut stripper,
                &mut suppressed_visible_fallback,
                &session_id,
                shell_kind,
                injection_sent_at.as_ref(),
                &suppression_diagnostics,
                InjectionTimeoutSource::WallClock,
                &pending_post_login,
                &mut post_login_deadline,
            );
        }

        tokio::select! {
            biased;

            _ = &mut inject_deadline,
                if phase == IoPhase::Suppressing && injection_sent_at.is_some() =>
            {
                fallback_shell_integration_timeout(
                    &mut phase,
                    &mut stripper,
                    &mut suppressed_visible_fallback,
                    &session_id,
                    shell_kind,
                    injection_sent_at.as_ref(),
                    &suppression_diagnostics,
                    InjectionTimeoutSource::Deadline,
                    &pending_post_login,
                    &mut post_login_deadline,
                );
            }
            cmd = cmd_rx.recv() => {
                match cmd {
                    Some(SessionCommand::AttachConfirmed { ack }) => {
                        output.attach_confirmed(ack);
                    }
                    Some(SessionCommand::DetachRenderer) => {
                        output.detach();
                    }
                    Some(SessionCommand::Write { data, raw, origin, .. }) => {
                        if zmodem_transfer.is_some()
                            || zmodem_upload_drain.should_suppress(std::time::Instant::now())
                        {
                            continue;
                        }
                        handle_input_before_initial_injection(
                            &mut phase,
                            &mut pending_script,
                            origin,
                            &manager,
                            &session_id,
                            &pending_post_login,
                            &mut post_login_deadline,
                        )
                        .await;
                        let send_data = prepare_terminal_write_input(data, &encoding, raw, backspace_as_bs);
                        if phase == IoPhase::Suppressing {
                            suppression_diagnostics.record_pre_ready_write(send_data.len());
                        }
                        let _ = channel.data(&send_data[..]).await;
                    }
                    Some(SessionCommand::Resize { cols, rows }) => {
                        let _ = channel.window_change(cols, rows, 0, 0).await;
                    }
                    Some(SessionCommand::PauseOutput) => {
                        output_paused = true;
                    }
                    Some(SessionCommand::ResumeOutput) => {
                        output_paused = false;
                    }
                    Some(SessionCommand::AckOutput { bytes }) => {
                        output.ack(bytes);
                    }
                    Some(SessionCommand::CaptureExec { marker_id, wrapped_command, result_tx }) => {
                        handle_input_before_initial_injection(
                            &mut phase,
                            &mut pending_script,
                            InputOrigin::AiAgent,
                            &manager,
                            &session_id,
                            &pending_post_login,
                            &mut post_login_deadline,
                        )
                        .await;
                        capture_processor.register(marker_id, result_tx);
                        let send_command = encode_terminal_input(&wrapped_command, &encoding);
                        let _ = channel.data(&send_command[..]).await;
                    }
                    Some(SessionCommand::CancelCapture { marker_id }) => {
                        capture_processor.cancel(&marker_id);
                    }
                    Some(SessionCommand::Close) => {
                        let _ = channel.close().await;
                        break "local-close-request";
                    }
                    Some(SessionCommand::ZmodemAcceptDownload { save_dir }) => {
                        if let Some(ref mut transfer) = zmodem_transfer {
                            let actions = transfer.accept_download(save_dir);
                            handle_zmodem_actions(&app, &zmodem_event_name, &mut channel, actions).await;
                            if transfer.is_done() {
                                zmodem_transfer = None;
                            }
                        }
                    }
                    Some(SessionCommand::ZmodemAcceptUpload {
                        files,
                        conflict_mode,
                        preserve_timestamps,
                    }) => {
                        if let Some(ref mut transfer) = zmodem_transfer {
                            let actions =
                                transfer.accept_upload(files, conflict_mode, preserve_timestamps);
                            handle_zmodem_actions(&app, &zmodem_event_name, &mut channel, actions).await;
                            if transfer.is_done() {
                                zmodem_transfer = None;
                            }
                        } else {
                            tracing::warn!(
                                session_id = %session_id,
                                "Received ZmodemAcceptUpload without an active transfer"
                            );
                        }
                    }
                    Some(SessionCommand::ZmodemCancel) => {
                        manager.clear_pending_zmodem_upload(&session_id).await;
                        if let Some(ref mut transfer) = zmodem_transfer {
                            let actions = transfer.cancel();
                            handle_zmodem_actions(&app, &zmodem_event_name, &mut channel, actions).await;
                        }
                        zmodem_transfer = None;
                    }
                    Some(SessionCommand::SerialModemUpload { result_tx, .. }) => {
                        let _ = result_tx.send(Err(
                            "Direct modem upload is only available for Serial sessions".to_string(),
                        ));
                    }
                    None => {
                        let _ = channel.close().await;
                        break "session-command-channel-closed";
                    }
                }
            }
            msg = channel.wait(), if !output_paused => {
                match msg {
                    Some(ChannelMsg::Data { ref data }) => {
                        if phase == IoPhase::Suppressing {
                            let now = Instant::now();
                            if let Some(sent_at) = injection_sent_at.as_ref() {
                                suppression_diagnostics.record_rx(data.len(), sent_at, now);
                                if injection_has_timed_out_at(&phase, Some(sent_at), now) {
                                    fallback_shell_integration_timeout(
                                        &mut phase,
                                        &mut stripper,
                                        &mut suppressed_visible_fallback,
                                        &session_id,
                                        shell_kind,
                                        injection_sent_at.as_ref(),
                                        &suppression_diagnostics,
                                        InjectionTimeoutSource::WallClock,
                                        &pending_post_login,
                                        &mut post_login_deadline,
                                    );
                                } else if suppression_diagnostics
                                    .should_log_suppression_diagnostic(sent_at, now)
                                {
                                    tracing::info!(
                                        session_id = %session_id,
                                        shell = ?shell_kind,
                                        elapsed_ms = elapsed_ms_at(sent_at, now),
                                        suppression_rx_bytes = suppression_diagnostics.rx_bytes_total,
                                        suppression_rx_chunks = suppression_diagnostics.rx_chunks,
                                        last_rx_after_ms = ?suppression_diagnostics.last_rx_after_ms,
                                        osc_buffer_bytes = stripper.buffered_len(),
                                        "SSH shell integration still suppressing"
                                    );
                                }
                            }
                        }
                        if !initial_remote_data_logged {
                            initial_remote_data_logged = true;
                            tracing::info!(
                                session_id = %session_id,
                                bytes = data.len(),
                                phase = %phase,
                                "SSH terminal received initial remote data"
                            );
                        }
                        // ZMODEM: if a transfer is active, route raw bytes to it.
                        if let Some(ref mut transfer) = zmodem_transfer {
                            let direction = transfer.direction();
                            let actions = transfer.feed_incoming(data);
                            handle_zmodem_actions(&app, &zmodem_event_name, &mut channel, actions).await;
                            if transfer.is_done() {
                                zmodem_transfer = None;
                                zmodem_detector.reset();
                                if direction == ZmodemDirection::Upload {
                                    zmodem_upload_drain.start(std::time::Instant::now());
                                } else if direction == ZmodemDirection::Download {
                                    zmodem_download_oo_drain.start(std::time::Instant::now());
                                }
                            }
                            continue;
                        }

                        let data = zmodem_upload_drain.filter(data, std::time::Instant::now());
                        if data.is_empty() {
                            continue;
                        }
                        let data = zmodem_download_oo_drain
                            .filter(data, std::time::Instant::now());
                        if data.is_empty() {
                            continue;
                        }

                        // ZMODEM: detect header in raw bytes before lossy UTF-8 conversion.
                        // Detection must run in every phase so an early `rz` is not missed
                        // while shell-integration output is still being suppressed.
                        match zmodem_detector.feed(data) {
                                ZmodemDetectResult::Detected { direction, passthrough, initial_bytes } => {
                                    // Forward any pre-header bytes to the terminal.
                                    if !passthrough.is_empty() {
                                        if let Some(ref recorder) = recording_mgr {
                                            recorder.write_raw_output(&session_id, &passthrough);
                                        }
                                        let pre = output_decoder.decode(&passthrough);
                                        if !pre.is_empty() {
                                            output.push_owned(pre);
                                        }
                                    }
                                    let prepared_upload = if direction == ZmodemDirection::Upload {
                                        manager.take_pending_zmodem_upload(&session_id).await
                                    } else {
                                        None
                                    };
                                    let (transfer, bootstrap_actions) =
                                        start_zmodem_transfer(direction, &initial_bytes, prepared_upload);
                                    zmodem_transfer = Some(transfer);
                                    handle_zmodem_actions(
                                        &app,
                                        &zmodem_event_name,
                                        &mut channel,
                                        bootstrap_actions,
                                    )
                                    .await;
                                    let _ = app.emit(&zmodem_event_name, &ZmodemEvent::Detected { direction });
                                    tracing::info!(
                                        session_id = %session_id,
                                        ?direction,
                                        "ZMODEM transfer detected"
                                    );
                                    continue;
                                }
                                ZmodemDetectResult::NoMatch { passthrough } if passthrough.is_empty() => {
                                    continue;
                                }
                                ZmodemDetectResult::NoMatch { passthrough } => {
                                    if let Some(ref recorder) = recording_mgr {
                                        recorder.write_raw_output(&session_id, &passthrough);
                                    }
                                    let text = output_decoder.decode(&passthrough);
                                    let mut result = stripper.push(&text);

                                    if capture_processor.has_active() {
                                        result.visible = capture_processor.process(&result.visible);
                                    }

                                    handle_osc_result(
                                        &app,
                                        &output,
                                        &cwd_event,
                                        &cwd,
                                        &recording_mgr,
                                        &session_id,
                                        &manager,
                                        &mut channel,
                                        &mut pending_script,
                                        &mut inject_deadline,
                                        &mut phase,
                                        &result,
                                        &mut suppressed_visible_fallback,
                                        shell_kind,
                                        &mut injection_sent_at,
                                        &mut suppression_diagnostics,
                                    ).await;
                                    arm_post_login_timer(
                                        &phase,
                                        &pending_post_login,
                                        &mut post_login_deadline,
                                    );
                                    continue;
                                }
                            }
                    }
                    Some(ChannelMsg::ExtendedData { ref data, .. }) => {
                        if let Some(ref recorder) = recording_mgr {
                            recorder.write_raw_output(&session_id, data);
                        }
                        let text = output_decoder.decode(data);
                        if let Some(ref recorder) = recording_mgr {
                            recorder.write_output(&session_id, &text);
                        }
                        output.push_owned(text);
                    }
                    Some(ChannelMsg::ExitStatus { exit_status }) => {
                        remote_exit_status = Some(exit_status);
                        tracing::info!(
                            session_id = %session_id,
                            exit_status,
                            "SSH remote process reported exit status"
                        );
                    }
                    Some(ChannelMsg::ExitSignal {
                        signal_name,
                        core_dumped,
                        error_message,
                        lang_tag,
                    }) => {
                        remote_exit_signal = Some(format!("{signal_name:?}"));
                        tracing::warn!(
                            session_id = %session_id,
                            signal = ?signal_name,
                            core_dumped,
                            error_message = %error_message,
                            lang_tag = %lang_tag,
                            "SSH remote process exited on signal"
                        );
                    }
                    Some(ChannelMsg::Eof) => {
                        tracing::info!(
                            session_id = %session_id,
                            "SSH channel received EOF from remote"
                        );
                        break "remote-channel-eof";
                    }
                    Some(ChannelMsg::Close) => {
                        tracing::info!(
                            session_id = %session_id,
                            "SSH channel received close from remote"
                        );
                        break "remote-channel-close";
                    }
                    None => {
                        tracing::info!(
                            session_id = %session_id,
                            "SSH channel stream ended"
                        );
                        break "channel-stream-ended";
                    }
                    _ => {}
                }
            }
            _ = &mut initial_inject_deadline, if should_send_initial_injection(&phase, pending_script.is_some()) => {
                send_shell_integration_injection(
                    &mut channel,
                    &mut pending_script,
                    &mut inject_deadline,
                    &mut phase,
                    &session_id,
                    shell_kind,
                    &mut injection_sent_at,
                    &mut suppression_diagnostics,
                ).await;
                arm_post_login_timer(&phase, &pending_post_login, &mut post_login_deadline);
            }
            _ = async {
                if let Some(deadline) = post_login_deadline.as_mut() {
                    deadline.as_mut().await;
                }
            }, if post_login_deadline.is_some() => {
                post_login_deadline = None;
                if zmodem_transfer.is_some() {
                    post_login_deadline = Some(Box::pin(tokio::time::sleep(
                        Duration::from_millis(250),
                    )));
                    continue;
                }
                if let Some(pending) = pending_post_login.take() {
                    let _ = channel.data(&pending.input[..]).await;
                    tracing::info!(
                        session_id = %session_id,
                        delay_ms = pending.delay_ms,
                        "Sent SSH post-login command"
                    );
                    arm_post_login_timer(
                        &phase,
                        &pending_startup_command,
                        &mut post_login_deadline,
                    );
                    continue;
                }
                if let Some(pending) = pending_startup_command.take() {
                    let _ = channel.data(&pending.input[..]).await;
                    tracing::info!(
                        session_id = %session_id,
                        delay_ms = pending.delay_ms,
                        "Sent SSH startup command"
                    );
                }
            }
        }
    };

    output.close();

    if let Some(ref recorder) = recording_mgr {
        recorder.cleanup_session(&session_id);
    }

    manager.remove_session(&session_id).await;
    if let Some(stats_sampler) = app.try_state::<Arc<RemoteStatsSampler>>() {
        stats_sampler.clear_session(&session_id).await;
    }

    if let Some(ref conn_id) = connection_id {
        if let Some(tunnel_mgr) = app.try_state::<Arc<super::TunnelManager>>() {
            tunnel_mgr
                .close_auto_tunnels_for_connection(&app, conn_id)
                .await;
        }
    }

    tracing::info!(
        session_id = %session_id,
        close_reason,
        remote_exit_status,
        remote_exit_signal = remote_exit_signal.as_deref(),
        "SSH session closed"
    );
    let _ = app.emit(&closed_event, ());
}

async fn run_sftp_only_session_commands(
    session_id: &str,
    manager: &Arc<SessionManager>,
    mut cmd_rx: SessionCommandReceiver,
    mut disconnect_rx: tokio::sync::mpsc::UnboundedReceiver<String>,
) -> &'static str {
    let close_reason = loop {
        tokio::select! {
            disconnect = disconnect_rx.recv() => {
                match disconnect {
                    Some(reason) => {
                        tracing::info!(
                            session_id = %session_id,
                            %reason,
                            "SFTP-only SSH transport disconnected"
                        );
                        break "remote-transport-disconnect";
                    }
                    None => break "remote-transport-ended",
                }
            }
            command = cmd_rx.recv() => match command {
                Some(SessionCommand::AttachConfirmed { ack }) => {
                    let _ = ack.send(());
                }
                Some(SessionCommand::CaptureExec { result_tx, .. }) => {
                    drop(result_tx);
                }
                Some(SessionCommand::SerialModemUpload { result_tx, .. }) => {
                    let _ = result_tx.send(Err(
                        "Direct modem upload is only available for Serial sessions".to_string(),
                    ));
                }
                Some(SessionCommand::Close) => break "local-close-request",
                Some(
                    SessionCommand::DetachRenderer
                    | SessionCommand::Write { .. }
                    | SessionCommand::PauseOutput
                    | SessionCommand::ResumeOutput
                    | SessionCommand::AckOutput { .. }
                    | SessionCommand::Resize { .. }
                    | SessionCommand::CancelCapture { .. }
                    | SessionCommand::ZmodemAcceptDownload { .. }
                    | SessionCommand::ZmodemAcceptUpload { .. }
                    | SessionCommand::ZmodemCancel,
                ) => {}
                None => break "session-command-channel-closed",
            }
        }
    };

    manager.remove_session(session_id).await;
    close_reason
}

pub(super) async fn sftp_only_ssh_lifecycle_loop(
    app: AppHandle,
    session_id: String,
    manager: Arc<SessionManager>,
    _handle: SshHandle,
    cmd_rx: SessionCommandReceiver,
    disconnect_rx: tokio::sync::mpsc::UnboundedReceiver<String>,
    connection_id: Option<String>,
) {
    tracing::info!(session_id = %session_id, "SFTP-only SSH lifecycle loop started");
    let close_reason =
        run_sftp_only_session_commands(&session_id, &manager, cmd_rx, disconnect_rx).await;

    if let Some(stats_sampler) = app.try_state::<Arc<RemoteStatsSampler>>() {
        stats_sampler.clear_session(&session_id).await;
    }

    if let Some(ref conn_id) = connection_id {
        if let Some(tunnel_mgr) = app.try_state::<Arc<super::TunnelManager>>() {
            tunnel_mgr
                .close_auto_tunnels_for_connection(&app, conn_id)
                .await;
        }
    }

    tracing::info!(
        session_id = %session_id,
        close_reason,
        "SFTP-only SSH session closed"
    );
    let closed_event = format!("session-closed-{session_id}");
    let _ = app.emit(&closed_event, ());
}

async fn handle_zmodem_actions(
    app: &AppHandle,
    event_name: &str,
    channel: &mut russh::Channel<client::Msg>,
    actions: Vec<ZmodemAction>,
) {
    for action in actions {
        match action {
            ZmodemAction::SendToRemote(data) => {
                let _ = channel.data(&data[..]).await;
            }
            ZmodemAction::EmitEvent(event) => {
                let _ = app.emit(event_name, &event);
            }
        }
    }
}

/// Helper: emit visible text + CWD updates from an [`OscResult`].
async fn emit_output(
    app: &AppHandle,
    output: &Arc<SessionOutputCoalescer>,
    cwd_event: &str,
    cwd: &SharedCwd,
    recording_mgr: &Option<Arc<RecordingManager>>,
    session_id: &str,
    manager: &Arc<SessionManager>,
    result: &osc::OscResult,
) {
    emit_metadata(app, cwd_event, cwd, manager, session_id, result).await;
    emit_visible_text(output, recording_mgr, session_id, &result.visible);
}

async fn emit_metadata(
    app: &AppHandle,
    cwd_event: &str,
    cwd: &SharedCwd,
    manager: &Arc<SessionManager>,
    session_id: &str,
    result: &osc::OscResult,
) {
    for path in &result.cwd_paths {
        if let Some(next_cwd) = update_cwd_if_changed(cwd, path).await {
            let _ = app.emit(cwd_event, &next_cwd);
        }
    }

    for command in &result.accepted_commands {
        let accepted = manager
            .confirm_command_submission(session_id, command.clone())
            .await;
        if accepted {
            let _ = app.emit(
                "session-command-accepted",
                serde_json::json!({
                    "sessionId": session_id,
                    "command": command,
                }),
            );
        }
    }
}

fn emit_visible_text(
    output: &Arc<SessionOutputCoalescer>,
    recording_mgr: &Option<Arc<RecordingManager>>,
    session_id: &str,
    visible: &str,
) {
    if visible.is_empty() {
        return;
    }

    if let Some(recorder) = recording_mgr {
        recorder.write_output(session_id, visible);
    }

    output.push(visible);
}

#[cfg(test)]
mod tests {
    use super::{
        INITIAL_INJECT_DELAY_MS, INJECT_TIMEOUT_SECS, InjectionEvent, InjectionTimeoutEvent,
        InjectionTimeoutSource, IoPhase, PendingStartupCommand,
        SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES, SUPPRESSION_DIAGNOSTIC_INITIAL_MS,
        SUPPRESSION_DIAGNOSTIC_INTERVAL_MS, SuppressionDiagnostics,
        append_suppressed_visible_and_take_passthrough, build_post_login_command_input,
        build_startup_command_input, cancel_pending_injection_for_input, discard_suppressed_output,
        fallback_shell_integration_timeout, handle_injection_result, handle_injection_timeout,
        injection_has_timed_out_at, on_initial_injection_sent, open_shell_channel,
        run_sftp_only_session_commands, should_send_initial_injection,
    };
    use crate::config::{AiExecutionProfile, SftpCwdFollowMode, SshProfile};
    use crate::core::InputOrigin;
    use crate::core::ssh::osc::{OscResult, OscStripper, build_ready_marker};
    use crate::core::{
        DynamicTitleCapabilities, SessionCommand, SessionHandle, SessionInfo, SessionManager,
        SessionType, session_command_channel,
    };
    use russh::{Channel, ChannelId, Disconnect, client, server};
    use std::pin::Pin;
    use std::sync::Arc;
    use std::time::Instant;
    use tokio::sync::{mpsc, oneshot};
    use tokio::time::{Duration, Sleep, timeout};

    struct TestClient;

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

        async fn check_server_key(
            &mut self,
            _server_public_key: &russh::keys::PublicKey,
        ) -> Result<bool, Self::Error> {
            Ok(true)
        }
    }

    fn assert_fake_server_session(result: Result<(), russh::Error>) {
        match result {
            Ok(()) => {}
            Err(russh::Error::IO(error)) if error.kind() == std::io::ErrorKind::BrokenPipe => {}
            Err(error) => panic!("fake SSH server session: {error:?}"),
        }
    }

    struct TestServer {
        agent_request_tx: mpsc::UnboundedSender<()>,
        reject_shell: bool,
        accept_x11: bool,
    }

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

        async fn auth_none(&mut self, _user: &str) -> Result<server::Auth, Self::Error> {
            Ok(server::Auth::Accept)
        }

        async fn channel_open_session(
            &mut self,
            _channel: Channel<server::Msg>,
            reply: server::ChannelOpenHandle,
            _session: &mut server::Session,
        ) -> Result<(), Self::Error> {
            reply.accept().await;
            Ok(())
        }

        async fn pty_request(
            &mut self,
            channel: ChannelId,
            _term: &str,
            _col_width: u32,
            _row_height: u32,
            _pix_width: u32,
            _pix_height: u32,
            _modes: &[(russh::Pty, u32)],
            session: &mut server::Session,
        ) -> Result<(), Self::Error> {
            let _ = session.channel_success(channel);
            Ok(())
        }

        async fn shell_request(
            &mut self,
            channel: ChannelId,
            session: &mut server::Session,
        ) -> Result<(), Self::Error> {
            if self.reject_shell {
                let _ = session.channel_failure(channel);
            } else {
                let _ = session.channel_success(channel);
            }
            Ok(())
        }

        async fn x11_request(
            &mut self,
            channel: ChannelId,
            _single_connection: bool,
            _x11_auth_protocol: &str,
            _x11_auth_cookie: &str,
            _x11_screen_number: u32,
            session: &mut server::Session,
        ) -> Result<(), Self::Error> {
            if self.accept_x11 {
                let _ = session.channel_success(channel);
            } else {
                let _ = session.channel_failure(channel);
            }
            Ok(())
        }

        async fn agent_request(
            &mut self,
            _channel: ChannelId,
            _session: &mut server::Session,
        ) -> Result<bool, Self::Error> {
            let _ = self.agent_request_tx.send(());
            Ok(true)
        }
    }

    async fn assert_agent_forwarding_request(enabled: bool) {
        let (client_stream, server_stream) = tokio::io::duplex(1024 * 1024);
        let (agent_request_tx, mut agent_request_rx) = mpsc::unbounded_channel();
        let mut rng = russh::keys::key::safe_rng();
        let server_config = Arc::new(server::Config {
            auth_rejection_time: Duration::ZERO,
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![
                russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)
                    .expect("test server key"),
            ],
            ..server::Config::default()
        });
        let server_task = tokio::spawn(async move {
            let session = server::run_stream(
                server_config,
                server_stream,
                TestServer {
                    agent_request_tx,
                    reject_shell: false,
                    accept_x11: true,
                },
            )
            .await
            .expect("fake SSH server handshake");
            assert_fake_server_session(session.await);
        });

        let mut handle = client::connect_stream(
            Arc::new(client::Config::default()),
            client_stream,
            TestClient,
        )
        .await
        .expect("fake SSH client handshake");
        assert!(
            handle
                .authenticate_none("test")
                .await
                .expect("none authentication")
                .success()
        );

        let (channel, ..) = open_shell_channel(
            &mut handle,
            "agent-forwarding-test",
            None,
            enabled,
            "xterm-256color",
            false,
            false,
            SftpCwdFollowMode::Off,
            100,
            None,
        )
        .await
        .expect("interactive shell channel");

        if enabled {
            timeout(Duration::from_secs(1), agent_request_rx.recv())
                .await
                .expect("fake server should receive Agent forwarding request")
                .expect("Agent forwarding request channel");
        } else {
            assert!(
                timeout(Duration::from_millis(100), agent_request_rx.recv())
                    .await
                    .is_err(),
                "disabled forwarding must not send auth-agent-req@openssh.com"
            );
        }

        drop(channel);
        handle
            .disconnect(Disconnect::ByApplication, "test complete", "")
            .await
            .expect("disconnect fake SSH session");
        timeout(Duration::from_secs(1), server_task)
            .await
            .expect("fake SSH server should stop")
            .expect("fake SSH server task");
    }

    #[tokio::test]
    async fn interactive_shell_rejection_is_reported() {
        let (client_stream, server_stream) = tokio::io::duplex(1024 * 1024);
        let (agent_request_tx, _agent_request_rx) = mpsc::unbounded_channel();
        let mut rng = russh::keys::key::safe_rng();
        let server_config = Arc::new(server::Config {
            auth_rejection_time: Duration::ZERO,
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![
                russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)
                    .expect("test server key"),
            ],
            ..server::Config::default()
        });
        let server_task = tokio::spawn(async move {
            let session = server::run_stream(
                server_config,
                server_stream,
                TestServer {
                    agent_request_tx,
                    reject_shell: true,
                    accept_x11: true,
                },
            )
            .await
            .expect("fake SSH server handshake");
            assert_fake_server_session(session.await);
        });

        let mut handle = client::connect_stream(
            Arc::new(client::Config::default()),
            client_stream,
            TestClient,
        )
        .await
        .expect("fake SSH client handshake");
        assert!(
            handle
                .authenticate_none("test")
                .await
                .expect("none authentication")
                .success()
        );

        let result = open_shell_channel(
            &mut handle,
            "shell-rejection-test",
            None,
            false,
            "xterm-256color",
            true,
            false,
            SftpCwdFollowMode::Off,
            100,
            None,
        )
        .await;
        let error = match result {
            Ok(_) => panic!("rejected shell request must fail interactive setup"),
            Err(error) => error,
        };
        assert!(
            error
                .to_string()
                .contains("Shell request rejected by server")
        );

        handle
            .disconnect(Disconnect::ByApplication, "test complete", "")
            .await
            .expect("disconnect fake SSH session");
        timeout(Duration::from_secs(1), server_task)
            .await
            .expect("fake SSH server should stop")
            .expect("fake SSH server task");
    }

    #[tokio::test]
    async fn interactive_shell_requests_agent_forwarding_only_when_enabled() {
        assert_agent_forwarding_request(false).await;
        assert_agent_forwarding_request(true).await;
    }

    #[tokio::test]
    async fn x11_success_does_not_hide_shell_rejection() {
        let (client_stream, server_stream) = tokio::io::duplex(1024 * 1024);
        let (agent_request_tx, _agent_request_rx) = mpsc::unbounded_channel();
        let mut rng = russh::keys::key::safe_rng();
        let server_config = Arc::new(server::Config {
            auth_rejection_time: Duration::ZERO,
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![
                russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)
                    .expect("test server key"),
            ],
            ..server::Config::default()
        });
        let server_task = tokio::spawn(async move {
            let session = server::run_stream(
                server_config,
                server_stream,
                TestServer {
                    agent_request_tx,
                    reject_shell: true,
                    accept_x11: true,
                },
            )
            .await
            .expect("fake SSH server handshake");
            assert_fake_server_session(session.await);
        });

        let mut handle = client::connect_stream(
            Arc::new(client::Config::default()),
            client_stream,
            TestClient,
        )
        .await
        .expect("fake SSH client handshake");
        assert!(
            handle
                .authenticate_none("test")
                .await
                .expect("none authentication")
                .success()
        );

        let error = open_shell_channel(
            &mut handle,
            "x11-shell-rejection-test",
            Some("00112233445566778899aabbccddeeff"),
            false,
            "xterm-256color",
            true,
            false,
            SftpCwdFollowMode::Off,
            100,
            None,
        )
        .await
        .expect_err("Shell rejection must not be consumed as an earlier X11 reply");
        assert!(
            error
                .to_string()
                .contains("Shell request rejected by server")
        );

        handle
            .disconnect(Disconnect::ByApplication, "test complete", "")
            .await
            .expect("disconnect fake SSH session");
        timeout(Duration::from_secs(1), server_task)
            .await
            .expect("fake SSH server should stop")
            .expect("fake SSH server task");
    }

    #[tokio::test]
    async fn x11_rejection_is_non_fatal_when_pty_and_shell_succeed() {
        let (client_stream, server_stream) = tokio::io::duplex(1024 * 1024);
        let (agent_request_tx, _agent_request_rx) = mpsc::unbounded_channel();
        let mut rng = russh::keys::key::safe_rng();
        let server_config = Arc::new(server::Config {
            auth_rejection_time: Duration::ZERO,
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![
                russh::keys::PrivateKey::random(&mut rng, russh::keys::Algorithm::Ed25519)
                    .expect("test server key"),
            ],
            ..server::Config::default()
        });
        let server_task = tokio::spawn(async move {
            let session = server::run_stream(
                server_config,
                server_stream,
                TestServer {
                    agent_request_tx,
                    reject_shell: false,
                    accept_x11: false,
                },
            )
            .await
            .expect("fake SSH server handshake");
            assert_fake_server_session(session.await);
        });

        let mut handle = client::connect_stream(
            Arc::new(client::Config::default()),
            client_stream,
            TestClient,
        )
        .await
        .expect("fake SSH client handshake");
        assert!(
            handle
                .authenticate_none("test")
                .await
                .expect("none authentication")
                .success()
        );

        let (channel, _, _, _, notice) = open_shell_channel(
            &mut handle,
            "x11-rejection-test",
            Some("00112233445566778899aabbccddeeff"),
            false,
            "xterm-256color",
            true,
            false,
            SftpCwdFollowMode::Off,
            100,
            None,
        )
        .await
        .expect("X11 rejection must not fail an otherwise valid interactive shell");
        assert!(notice.is_some());

        drop(channel);
        handle
            .disconnect(Disconnect::ByApplication, "test complete", "")
            .await
            .expect("disconnect fake SSH session");
        timeout(Duration::from_secs(1), server_task)
            .await
            .expect("fake SSH server should stop")
            .expect("fake SSH server task");
    }

    #[tokio::test]
    async fn sftp_only_session_close_cleans_up() {
        let session_id = "sftp-only-session".to_string();
        let manager = Arc::new(SessionManager::new());
        let (cmd_tx, cmd_rx) = session_command_channel(session_id.clone());
        let info = SessionInfo {
            id: session_id.clone(),
            name: "sftp-only".to_string(),
            session_type: SessionType::SSH,
            started_at: String::new(),
            connection_id: None,
            connected: true,
            owner_window_label: None,
            ai_execution_profile: AiExecutionProfile::Disabled,
            injection_active: false,
            dynamic_title_capabilities: DynamicTitleCapabilities::default(),
            remote_file_browser_enabled: true,
            remote_stats_enabled: false,
            ssh_profile: Some(SshProfile::Standard),
            ssh_runtime_mode: Some(crate::config::SshRuntimeMode::Sftp),
        };
        manager
            .add_session(SessionHandle {
                info,
                cmd_tx,
                startup_input_barrier: None,
                ssh_config: None,
                ssh_handle: None,
                cwd: Arc::new(tokio::sync::Mutex::new(Default::default())),
                remote_fs: None,
            })
            .await;

        let loop_manager = manager.clone();
        let loop_session_id = session_id.clone();
        let (_disconnect_tx, disconnect_rx) = mpsc::unbounded_channel();
        let loop_task = tokio::spawn(async move {
            run_sftp_only_session_commands(&loop_session_id, &loop_manager, cmd_rx, disconnect_rx)
                .await
        });

        let (attach_tx, attach_rx) = oneshot::channel();
        manager
            .send_command(
                &session_id,
                SessionCommand::AttachConfirmed { ack: attach_tx },
            )
            .await
            .expect("attach command");
        timeout(Duration::from_secs(1), attach_rx)
            .await
            .expect("attach acknowledgement must not block")
            .expect("attach acknowledgement sender");

        let (capture_tx, capture_rx) = oneshot::channel();
        manager
            .send_command(
                &session_id,
                SessionCommand::CaptureExec {
                    marker_id: "capture".to_string(),
                    wrapped_command: b"echo blocked".to_vec(),
                    result_tx: capture_tx,
                },
            )
            .await
            .expect("capture command");
        let capture_result = timeout(Duration::from_secs(1), capture_rx)
            .await
            .expect("capture result sender must be dropped immediately");
        assert!(capture_result.is_err());

        manager
            .send_command(&session_id, SessionCommand::Close)
            .await
            .expect("close command");
        let close_reason = timeout(Duration::from_secs(1), loop_task)
            .await
            .expect("SFTP-only loop must close promptly")
            .expect("SFTP-only loop task");

        assert_eq!(close_reason, "local-close-request");
        assert!(manager.list_sessions().await.is_empty());
    }

    #[tokio::test]
    async fn sftp_only_session_remote_disconnect_cleans_up() {
        let session_id = "sftp-only-remote-disconnect".to_string();
        let manager = Arc::new(SessionManager::new());
        let (cmd_tx, cmd_rx) = session_command_channel(session_id.clone());
        let info = SessionInfo {
            id: session_id.clone(),
            name: "sftp-only".to_string(),
            session_type: SessionType::SSH,
            started_at: String::new(),
            connection_id: None,
            connected: true,
            owner_window_label: None,
            ai_execution_profile: AiExecutionProfile::Disabled,
            injection_active: false,
            dynamic_title_capabilities: DynamicTitleCapabilities::default(),
            remote_file_browser_enabled: true,
            remote_stats_enabled: false,
            ssh_profile: Some(SshProfile::Standard),
            ssh_runtime_mode: Some(crate::config::SshRuntimeMode::Sftp),
        };
        manager
            .add_session(SessionHandle {
                info,
                cmd_tx,
                startup_input_barrier: None,
                ssh_config: None,
                ssh_handle: None,
                cwd: Arc::new(tokio::sync::Mutex::new(Default::default())),
                remote_fs: None,
            })
            .await;

        let (disconnect_tx, disconnect_rx) = mpsc::unbounded_channel();
        let loop_manager = manager.clone();
        let loop_session_id = session_id.clone();
        let loop_task = tokio::spawn(async move {
            run_sftp_only_session_commands(&loop_session_id, &loop_manager, cmd_rx, disconnect_rx)
                .await
        });

        disconnect_tx
            .send("SSH server disconnected: test".to_string())
            .expect("disconnect notification");
        let close_reason = timeout(Duration::from_secs(1), loop_task)
            .await
            .expect("SFTP-only loop must react to remote disconnect")
            .expect("SFTP-only loop task");

        assert_eq!(close_reason, "remote-transport-disconnect");
        assert!(manager.list_sessions().await.is_empty());
    }

    #[test]
    fn post_login_input_normalizes_line_endings_and_adds_enter() {
        let input = build_post_login_command_input("cd /opt/app\nclear").expect("input");

        assert_eq!(input, b"cd /opt/app\rclear\r");
    }

    #[test]
    fn post_login_input_preserves_existing_trailing_enter() {
        let input = build_post_login_command_input("uptime\r").expect("input");

        assert_eq!(input, b"uptime\r");
    }

    #[test]
    fn post_login_input_ignores_blank_commands() {
        assert!(build_post_login_command_input(" \n\t ").is_none());
    }

    #[test]
    fn startup_input_rejects_c0_del_and_c1_controls() {
        for code in 0x00..=0x1f {
            let command = format!("cd /tmp/a{}b", char::from_u32(code).expect("C0 code point"));
            assert!(
                build_startup_command_input(&command).is_none(),
                "C0 control U+{code:04X} must be rejected"
            );
        }
        for code in 0x7f..=0x9f {
            let command = format!("cd /tmp/a{}b", char::from_u32(code).expect("C1 code point"));
            assert!(
                build_startup_command_input(&command).is_none(),
                "DEL/C1 control U+{code:04X} must be rejected"
            );
        }
    }

    #[test]
    fn startup_input_adds_enter_to_a_safe_command() {
        let input = build_startup_command_input("cd '/opt/app'").expect("input");

        assert_eq!(input, b"cd '/opt/app'\r");
    }

    fn osc_result(ready: bool, visible_after_ready: &str) -> OscResult {
        OscResult {
            visible: "suppressed output".to_string(),
            visible_after_ready: visible_after_ready.to_string(),
            cwd_paths: Vec::new(),
            cwd_payloads: Vec::new(),
            cwd_payload_invalidated: false,
            cwd_payload_events: Vec::new(),
            ready,
            ready_failed: false,
            accepted_commands: Vec::new(),
        }
    }

    #[test]
    fn initial_inject_delay_is_500ms_and_wait_initial_can_inject_without_output() {
        assert_eq!(INITIAL_INJECT_DELAY_MS, 500);
        assert!(should_send_initial_injection(&IoPhase::WaitInitial, true));
        assert!(!should_send_initial_injection(&IoPhase::WaitInitial, false));
        assert!(!should_send_initial_injection(&IoPhase::Suppressing, true));
        assert!(!should_send_initial_injection(&IoPhase::Normal, true));

        let mut phase = IoPhase::WaitInitial;
        assert_eq!(
            handle_injection_result(&mut phase, &osc_result(false, "")),
            InjectionEvent::Inject
        );
        assert_eq!(phase, IoPhase::WaitInitial);
        on_initial_injection_sent(&mut phase);
        assert_eq!(phase, IoPhase::Suppressing);
    }

    #[test]
    fn real_input_before_initial_injection_cancels_pending_script_and_preserves_output() {
        let ready_marker = build_ready_marker("session-1");
        let mut stripper = OscStripper::new(&ready_marker);
        let mut phase = IoPhase::WaitInitial;
        let mut pending_script =
            Some("source ~/.config/nyaterm/shell-integration.bash\n".to_string());

        assert!(cancel_pending_injection_for_input(
            &mut phase,
            &mut pending_script,
            InputOrigin::Keyboard,
        ));
        assert_eq!(phase, IoPhase::Normal);
        assert!(pending_script.is_none());

        let chunks = [
            "\x1b[38;5;14mubuntu-logo\x1b[0m\nOS: Ubuntu 22.04\n",
            "Kernel: 6.8.0\nUptime: 1 day\n",
        ];
        let mut visible = String::new();
        for chunk in chunks {
            let result = stripper.push(chunk);
            assert_eq!(
                handle_injection_result(&mut phase, &result),
                InjectionEvent::None
            );
            assert_eq!(phase, IoPhase::Normal);
            visible.push_str(&result.visible);
        }

        assert_eq!(visible, chunks.concat());
    }

    #[test]
    fn terminal_response_keeps_pending_initial_injection() {
        let mut phase = IoPhase::WaitInitial;
        let mut pending_script = Some("integration-script".to_string());

        assert!(!cancel_pending_injection_for_input(
            &mut phase,
            &mut pending_script,
            InputOrigin::TerminalResponse,
        ));
        assert_eq!(phase, IoPhase::WaitInitial);
        assert!(pending_script.is_some());
        assert_eq!(
            handle_injection_result(&mut phase, &osc_result(false, "")),
            InjectionEvent::Inject
        );
        on_initial_injection_sent(&mut phase);
        assert_eq!(phase, IoPhase::Suppressing);
    }

    #[test]
    fn input_after_initial_injection_does_not_end_suppression() {
        let ready_marker = build_ready_marker("session-1");
        let mut stripper = OscStripper::new(&ready_marker);
        let mut phase = IoPhase::Suppressing;
        let mut pending_script = None;

        assert!(!cancel_pending_injection_for_input(
            &mut phase,
            &mut pending_script,
            InputOrigin::Keyboard,
        ));
        let source_echo = stripper.push("hidden integration echo");
        assert_eq!(
            handle_injection_result(&mut phase, &source_echo),
            InjectionEvent::None
        );
        assert_eq!(phase, IoPhase::Suppressing);

        let ready = stripper.push(&ready_marker);
        assert!(matches!(
            handle_injection_result(&mut phase, &ready),
            InjectionEvent::Ready { .. }
        ));
        assert_eq!(phase, IoPhase::Normal);
    }

    #[test]
    fn ready_marker_in_suppressing_enters_normal_and_preserves_prompt_after_ready() {
        let mut phase = IoPhase::Suppressing;
        let sent_at = Instant::now();
        assert!(!injection_has_timed_out_at(
            &phase,
            Some(&sent_at),
            sent_at + Duration::from_millis(7_900)
        ));
        let result = osc_result(true, "[user@host ~]$ ");

        let event = handle_injection_result(&mut phase, &result);

        assert_eq!(phase, IoPhase::Normal);
        assert_eq!(
            event,
            InjectionEvent::Ready {
                visible_after_ready: "[user@host ~]$ ".to_string()
            }
        );
    }

    #[test]
    fn failure_marker_in_suppressing_enters_normal_and_preserves_prompt() {
        let mut phase = IoPhase::Suppressing;
        let mut result = osc_result(false, "prompt-after-failure");
        result.ready_failed = true;

        let event = handle_injection_result(&mut phase, &result);

        assert_eq!(phase, IoPhase::Normal);
        assert_eq!(
            event,
            InjectionEvent::Failed {
                visible_after_ready: "prompt-after-failure".to_string()
            }
        );
    }

    #[test]
    fn failure_marker_after_startup_preserves_visible_output_and_reports_runtime_failure() {
        let mut phase = IoPhase::Normal;
        let mut result = osc_result(false, "prompt-after-failure");
        result.visible = "beforeprompt-after-failure".to_string();
        result.ready_failed = true;

        let event = handle_injection_result(&mut phase, &result);

        assert_eq!(phase, IoPhase::Normal);
        assert_eq!(
            event,
            InjectionEvent::RuntimeFailed {
                visible: "beforeprompt-after-failure".to_string()
            }
        );
    }

    #[test]
    fn suppression_transport_preserves_interleaved_title_query_order() {
        let mut buffer = String::new();
        let controls = concat!(
            "\x1b]2;first\x07",
            "\x1b[c",
            "\x1b]0;final\x1b\\",
            "\x1b]11;?\x07"
        );

        let passthrough = append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &format!("hidden{controls}tail"),
        );

        assert_eq!(passthrough, controls);
        assert_eq!(buffer, "hiddentail");
    }

    #[test]
    fn oversized_startup_queries_and_titles_remain_suppressed() {
        let mut buffer = String::new();
        let huge_dcs = format!("\x1bP$q{}\x1b\\", "x".repeat(17 * 1024));
        let huge_title = format!("\x1b]2;{}\x07", "x".repeat(17 * 1024));

        let passthrough = append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &format!("{huge_dcs}{huge_title}"),
        );

        assert!(passthrough.is_empty());
        assert!(buffer.len() <= SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES);
    }

    #[test]
    fn suppression_title_transport_is_exactly_once_across_split_ready_and_normal_phases() {
        let ready_marker = build_ready_marker("session-1");
        let mut stripper = OscStripper::new(&ready_marker);
        let mut phase = IoPhase::Suppressing;
        let mut buffer = String::new();
        let mut emitted = String::new();

        let incomplete = stripper.push("hidden source\x1b]2;split");
        let _ = handle_injection_result(&mut phase, &incomplete);
        emitted.push_str(&append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &incomplete.visible,
        ));

        let completed = stripper.push(" title\x07hidden tail");
        let _ = handle_injection_result(&mut phase, &completed);
        emitted.push_str(&append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &completed.visible,
        ));

        let ready = stripper.push(&format!("{ready_marker}[user@host]$ "));
        let event = handle_injection_result(&mut phase, &ready);
        let before_ready_len = ready.visible.len() - ready.visible_after_ready.len();
        emitted.push_str(&append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &ready.visible[..before_ready_len],
        ));
        assert!(matches!(event, InjectionEvent::Ready { .. }));
        emitted.push_str(&ready.visible_after_ready);
        buffer.clear();

        let normal = stripper.push("\x1b]2;normal title\x07prompt");
        let event = handle_injection_result(&mut phase, &normal);
        assert_eq!(event, InjectionEvent::None);
        emitted.push_str(&normal.visible);

        assert_eq!(phase, IoPhase::Normal);
        assert_eq!(emitted.matches("\x1b]2;split title\x07").count(), 1);
        assert_eq!(emitted.matches("\x1b]2;normal title\x07").count(), 1);
        assert!(!emitted.contains("hidden source"));
        assert!(!emitted.contains("hidden tail"));
        assert!(
            emitted.find("\x1b]2;split title\x07").unwrap()
                < emitted.find("[user@host]$ ").unwrap()
        );
        assert!(
            emitted.find("[user@host]$ ").unwrap()
                < emitted.find("\x1b]2;normal title\x07").unwrap()
        );
    }

    #[test]
    fn title_transport_cap_is_per_sequence_and_keeps_final_semantics() {
        let first = format!("\x1b]2;{}\x07", "x".repeat(16_395));
        let final_title = "\x1b]2;final\x07";
        let mut buffer = String::new();

        let passthrough = append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &format!("{first}{final_title}"),
        );

        assert_eq!(passthrough, format!("{first}{final_title}"));
        assert!(passthrough.ends_with(final_title));
        assert!(buffer.is_empty());
    }

    #[test]
    fn delivered_suppression_title_survives_timeout_fallback() {
        let mut buffer = String::new();
        let delivered = append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            "\x1b]2;delivered\x07hidden",
        );
        let mut phase = IoPhase::Suppressing;

        assert_eq!(
            handle_injection_timeout(&mut phase),
            InjectionTimeoutEvent::FallbackToNormal
        );
        assert_eq!(delivered, "\x1b]2;delivered\x07");
        assert_eq!(buffer, "hidden");
        assert_eq!(phase, IoPhase::Normal);
    }

    #[test]
    fn injection_timeout_is_8s_and_only_applies_after_injection_is_sent() {
        assert_eq!(INJECT_TIMEOUT_SECS, 8);

        let mut wait_initial = IoPhase::WaitInitial;
        assert_eq!(
            handle_injection_timeout(&mut wait_initial),
            InjectionTimeoutEvent::None
        );
        assert_eq!(wait_initial, IoPhase::WaitInitial);
        let sent_at = Instant::now();
        assert!(!injection_has_timed_out_at(
            &wait_initial,
            Some(&sent_at),
            sent_at + Duration::from_secs(INJECT_TIMEOUT_SECS)
        ));

        let mut suppressing = IoPhase::Suppressing;
        assert!(!injection_has_timed_out_at(
            &suppressing,
            None,
            sent_at + Duration::from_secs(INJECT_TIMEOUT_SECS)
        ));
        assert!(injection_has_timed_out_at(
            &suppressing,
            Some(&sent_at),
            sent_at + Duration::from_secs(INJECT_TIMEOUT_SECS)
        ));
        assert_eq!(
            handle_injection_timeout(&mut suppressing),
            InjectionTimeoutEvent::FallbackToNormal
        );
        assert_eq!(suppressing, IoPhase::Normal);
    }

    #[test]
    fn continuous_remote_data_cannot_extend_wall_clock_timeout() {
        let sent_at = Instant::now();
        let mut phase = IoPhase::Suppressing;
        let mut diagnostics = SuppressionDiagnostics::default();

        for chunk in 0..80 {
            let now = sent_at + Duration::from_millis(chunk * 100);
            diagnostics.record_rx(4, &sent_at, now);
            assert!(!injection_has_timed_out_at(&phase, Some(&sent_at), now));
        }

        let timeout_at = sent_at + Duration::from_secs(INJECT_TIMEOUT_SECS);
        diagnostics.record_rx(4, &sent_at, timeout_at);
        assert!(injection_has_timed_out_at(
            &phase,
            Some(&sent_at),
            timeout_at
        ));
        assert_eq!(diagnostics.rx_bytes_total, 324);
        assert_eq!(diagnostics.rx_chunks, 81);
        assert_eq!(diagnostics.first_rx_after_ms, Some(0));
        assert_eq!(diagnostics.last_rx_after_ms, Some(8_000));
        assert_eq!(
            handle_injection_timeout(&mut phase),
            InjectionTimeoutEvent::FallbackToNormal
        );
        assert_eq!(phase, IoPhase::Normal);
    }

    #[tokio::test]
    async fn expired_injection_deadline_wins_over_ready_remote_data() {
        let deadline = tokio::time::sleep(Duration::ZERO);
        tokio::pin!(deadline);
        let remote_data = std::future::ready(());

        let selected = tokio::select! {
            biased;
            () = &mut deadline => "deadline",
            () = remote_data => "remote_data",
        };

        assert_eq!(selected, "deadline");
    }

    #[test]
    fn ready_marker_after_timeout_stays_normal_and_is_safely_stripped() {
        let ready_marker = build_ready_marker("session-1");
        let mut stripper = OscStripper::new(&ready_marker);
        let mut phase = IoPhase::Suppressing;
        assert_eq!(
            handle_injection_timeout(&mut phase),
            InjectionTimeoutEvent::FallbackToNormal
        );

        let result = stripper.push(&format!("{ready_marker}late prompt"));
        let event = handle_injection_result(&mut phase, &result);

        assert_eq!(phase, IoPhase::Normal);
        assert_eq!(event, InjectionEvent::None);
        assert_eq!(result.visible, "late prompt");
        assert!(result.ready);
    }

    #[test]
    fn pre_ready_writes_are_counted_without_extending_timeout() {
        let sent_at = Instant::now();
        let phase = IoPhase::Suppressing;
        let mut diagnostics = SuppressionDiagnostics::default();

        diagnostics.record_pre_ready_write(3);
        diagnostics.record_pre_ready_write(5);

        assert_eq!(diagnostics.pre_ready_write_bytes, 8);
        assert_eq!(diagnostics.pre_ready_write_chunks, 2);
        assert!(injection_has_timed_out_at(
            &phase,
            Some(&sent_at),
            sent_at + Duration::from_secs(INJECT_TIMEOUT_SECS)
        ));
    }

    #[test]
    fn suppression_diagnostic_is_delayed_and_rate_limited() {
        assert_eq!(SUPPRESSION_DIAGNOSTIC_INITIAL_MS, 1_000);
        assert_eq!(SUPPRESSION_DIAGNOSTIC_INTERVAL_MS, 2_000);
        let sent_at = Instant::now();
        let mut diagnostics = SuppressionDiagnostics::default();

        assert!(
            !diagnostics
                .should_log_suppression_diagnostic(&sent_at, sent_at + Duration::from_millis(999))
        );
        assert!(
            diagnostics
                .should_log_suppression_diagnostic(&sent_at, sent_at + Duration::from_secs(1))
        );
        assert!(
            !diagnostics.should_log_suppression_diagnostic(
                &sent_at,
                sent_at + Duration::from_millis(2_999)
            )
        );
        assert!(
            diagnostics
                .should_log_suppression_diagnostic(&sent_at, sent_at + Duration::from_secs(3))
        );
    }

    #[tokio::test]
    async fn timeout_fallback_is_idempotent_flushes_buffers_and_arms_post_login() {
        let ready_marker = build_ready_marker("session-1");
        let mut stripper = OscStripper::new(&ready_marker);
        let _ = stripper.push("\x1b]2;unfinished");
        let mut suppressed_visible = "hidden startup output".to_string();
        let mut phase = IoPhase::Suppressing;
        let sent_at = Instant::now()
            .checked_sub(Duration::from_secs(INJECT_TIMEOUT_SECS))
            .expect("test instant should support an eight-second subtraction");
        let diagnostics = SuppressionDiagnostics::default();
        let pending_post_login = Some(PendingStartupCommand {
            input: b"uptime\r".to_vec(),
            delay_ms: 1,
        });
        let mut post_login_deadline: Option<Pin<Box<Sleep>>> = None;

        assert!(fallback_shell_integration_timeout(
            &mut phase,
            &mut stripper,
            &mut suppressed_visible,
            "session-1",
            Some(crate::core::ssh::osc::ShellKind::Bash),
            Some(&sent_at),
            &diagnostics,
            InjectionTimeoutSource::Deadline,
            &pending_post_login,
            &mut post_login_deadline,
        ));
        assert_eq!(phase, IoPhase::Normal);
        assert!(suppressed_visible.is_empty());
        assert_eq!(stripper.buffered_len(), 0);
        assert!(post_login_deadline.is_some());

        assert!(!fallback_shell_integration_timeout(
            &mut phase,
            &mut stripper,
            &mut suppressed_visible,
            "session-1",
            Some(crate::core::ssh::osc::ShellKind::Bash),
            Some(&sent_at),
            &diagnostics,
            InjectionTimeoutSource::WallClock,
            &pending_post_login,
            &mut post_login_deadline,
        ));
    }

    #[test]
    fn suppressed_visible_fallback_keeps_recent_text_and_extracts_split_passthrough() {
        let mut buffer = String::new();

        assert!(
            append_suppressed_visible_and_take_passthrough(&mut buffer, "hidden\x1b[").is_empty()
        );
        let passthrough =
            append_suppressed_visible_and_take_passthrough(&mut buffer, "cstill-hidden");

        assert_eq!(passthrough, "\x1b[c");
        assert_eq!(buffer, "hiddenstill-hidden");
    }

    #[test]
    fn suppressed_visible_fallback_is_bounded_after_passthrough_extraction() {
        let mut buffer = String::new();

        let passthrough = append_suppressed_visible_and_take_passthrough(
            &mut buffer,
            &format!(
                "{}\x1b]11;?\x07",
                "x".repeat(SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES + 16)
            ),
        );

        assert_eq!(passthrough, "\x1b]11;?\x07");
        assert_eq!(buffer.len(), SUPPRESSED_VISIBLE_FALLBACK_MAX_BYTES);
    }

    #[test]
    fn timeout_discard_clears_suppressed_visible_and_buffered_osc_text() {
        let mut buffer = "prompt".to_string();

        let discarded_len = discard_suppressed_output(&mut buffer, "tail".to_string());

        assert_eq!(discarded_len, "prompttail".len());
        assert!(buffer.is_empty());
    }

    #[tokio::test]
    async fn post_login_timer_only_arms_after_injection_is_normal() {
        let pending_post_login = Some(PendingStartupCommand {
            input: b"uptime\r".to_vec(),
            delay_ms: 1,
        });
        let mut post_login_deadline: Option<Pin<Box<Sleep>>> = None;

        super::arm_post_login_timer(
            &IoPhase::WaitInitial,
            &pending_post_login,
            &mut post_login_deadline,
        );
        assert!(post_login_deadline.is_none());

        super::arm_post_login_timer(
            &IoPhase::Suppressing,
            &pending_post_login,
            &mut post_login_deadline,
        );
        assert!(post_login_deadline.is_none());

        super::arm_post_login_timer(
            &IoPhase::Normal,
            &pending_post_login,
            &mut post_login_deadline,
        );

        assert!(post_login_deadline.is_some());
        post_login_deadline.as_mut().unwrap().as_mut().await;
    }

    #[tokio::test]
    async fn early_input_cancellation_arms_post_login_timer() {
        let pending_post_login = Some(PendingStartupCommand {
            input: b"uptime\r".to_vec(),
            delay_ms: 1,
        });
        let mut pending_script = Some("integration-script".to_string());
        let mut phase = IoPhase::WaitInitial;
        let mut post_login_deadline: Option<Pin<Box<Sleep>>> = None;

        assert!(cancel_pending_injection_for_input(
            &mut phase,
            &mut pending_script,
            InputOrigin::QuickCommand,
        ));
        super::arm_post_login_timer(&phase, &pending_post_login, &mut post_login_deadline);

        assert_eq!(phase, IoPhase::Normal);
        assert!(post_login_deadline.is_some());
        post_login_deadline.as_mut().unwrap().as_mut().await;
    }

    #[tokio::test]
    async fn capture_exec_before_initial_injection_cancels_pending_script() {
        let pending_post_login = Some(PendingStartupCommand {
            input: b"uptime\r".to_vec(),
            delay_ms: 1,
        });
        let mut pending_script = Some("integration-script".to_string());
        let mut phase = IoPhase::WaitInitial;
        let mut post_login_deadline: Option<Pin<Box<Sleep>>> = None;

        assert!(cancel_pending_injection_for_input(
            &mut phase,
            &mut pending_script,
            InputOrigin::AiAgent,
        ));
        super::arm_post_login_timer(&phase, &pending_post_login, &mut post_login_deadline);

        assert_eq!(phase, IoPhase::Normal);
        assert!(pending_script.is_none());
        assert!(post_login_deadline.is_some());
        post_login_deadline.as_mut().unwrap().as_mut().await;
    }
}