//! SSH session manager (port of `sessions/sessionManager.js`, connection part).
//!
//! One SSH connection per session is shared by the terminal, SFTP, metrics, Users, NGINX,
//! disks and swap services. Host keys are verified against the SHA-256 fingerprints stored
//! in the vault (same hex format as ssh2 `hostHash: "sha256"` in 0.1.0); unknown or changed
//! keys are refused and kept as a 5-minute pending confirmation.

use crate::core::availability::Outcome;
use crate::core::error::{CoreError, CoreResult};
use crate::core::logging::Logger;
use crate::core::profiles::{normalize_profile, ProfileService};
use crate::core::ssh::exec::{ExecChannel, ExecResult, MAX_COMMAND_OUTPUT};
use crate::core::ssh::keys::{decode_private_key, read_private_key_file, sha256_hex};
use crate::core::ssh::utf8::Utf8Decoder;
use once_cell::sync::Lazy;
use parking_lot::Mutex;
use regex::Regex;
use russh::client::{self, Handle, Msg};
use russh::keys::{PrivateKeyWithHashAlg, PublicKeyOrCertificate};
use russh::{ChannelMsg, ChannelWriteHalf};
use serde_json::{json, Value};
use std::collections::HashMap;
use std::net::{IpAddr, SocketAddr};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{broadcast, mpsc, watch};
use zeroize::Zeroizing;

static SESSION_ID: Lazy<Regex> = Lazy::new(|| Regex::new(r"^[A-Za-z0-9._-]{1,80}$").unwrap());
pub const MAX_INPUT_LENGTH: usize = 64 * 1024;
const READY_TIMEOUT: Duration = Duration::from_secs(15);
const PENDING_TRUST_TTL: Duration = Duration::from_secs(5 * 60);

pub fn normalize_terminal_size(value: Option<&Value>, fallback: u32, min: u32, max: u32) -> u32 {
    match value.and_then(Value::as_f64) {
        Some(n) if n.fract() == 0.0 => (n.max(min as f64).min(max as f64)) as u32,
        _ => fallback,
    }
}

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

// ---------------------------------------------------------------------------------------------
// Client handler: host-key verification
// ---------------------------------------------------------------------------------------------

pub struct ClientHandler {
    expected: Option<String>,
    observed: Arc<Mutex<Option<String>>>,
}

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

    async fn check_server_key(&mut self, key: &PublicKeyOrCertificate) -> Result<bool, Self::Error> {
        let blob = match key {
            PublicKeyOrCertificate::PublicKey { key, .. } => key.to_bytes()?,
            PublicKeyOrCertificate::Certificate(certificate) => certificate.to_bytes()?,
        };
        let fingerprint = sha256_hex(&blob);
        *self.observed.lock() = Some(fingerprint.clone());
        Ok(self.expected.as_deref() == Some(fingerprint.as_str()))
    }
}

/// Algorithm lists of ssh2 1.17 defaults used by 0.1.0 (plus ML-KEM hybrid KEX first).
fn client_config(keepalive: bool) -> Arc<client::Config> {
    use russh::{cipher, kex, mac, Preferred};
    use std::borrow::Cow;
    let preferred = Preferred {
        kex: Cow::Owned(vec![
            kex::MLKEM768X25519_SHA256,
            kex::CURVE25519,
            kex::CURVE25519_PRE_RFC_8731,
            kex::ECDH_SHA2_NISTP256,
            kex::ECDH_SHA2_NISTP384,
            kex::ECDH_SHA2_NISTP521,
            kex::DH_GEX_SHA256,
            kex::DH_G14_SHA256,
            kex::DH_G15_SHA512,
            kex::DH_G16_SHA512,
            kex::DH_G17_SHA512,
            kex::DH_G18_SHA512,
            kex::EXTENSION_SUPPORT_AS_CLIENT,
            kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT,
        ]),
        cipher: Cow::Owned(vec![
            cipher::CHACHA20_POLY1305,
            cipher::AES_128_GCM,
            cipher::AES_256_GCM,
            cipher::AES_128_CTR,
            cipher::AES_192_CTR,
            cipher::AES_256_CTR,
        ]),
        mac: Cow::Owned(vec![
            mac::HMAC_SHA256_ETM,
            mac::HMAC_SHA512_ETM,
            mac::HMAC_SHA1_ETM,
            mac::HMAC_SHA256,
            mac::HMAC_SHA512,
            mac::HMAC_SHA1,
        ]),
        ..Preferred::DEFAULT
    };
    Arc::new(client::Config {
        preferred,
        inactivity_timeout: None,
        keepalive_interval: keepalive.then(|| Duration::from_secs(15)),
        keepalive_max: 3,
        nodelay: true,
        ..Default::default()
    })
}

// ---------------------------------------------------------------------------------------------
// Session records
// ---------------------------------------------------------------------------------------------

enum ShellCommand {
    Input(Vec<u8>),
    Resize(u32, u32),
}

pub struct SessionRecord {
    pub session_id: String,
    pub profile_id: String,
    pub label: String,
    pub host: String,
    pub address: Mutex<String>,
    pub port: u16,
    pub username: String,
    pub auth_method: String,
    pub slot: Option<u8>,
    pub adhoc: bool,
    profile: Value,
    route: Vec<Value>,
    jump_credentials: Mutex<Option<Vec<Credentials>>>,
    transport: Mutex<Option<OpenedRoute>>,
    pub state: Mutex<String>,
    pub handle: Mutex<Option<Arc<Handle<ClientHandler>>>>,
    shell: Mutex<Option<mpsc::UnboundedSender<ShellCommand>>>,
    pub latency: Mutex<Option<Value>>,
    pub finalized: AtomicBool,
    lifecycle: Mutex<()>,
    lost_tx: watch::Sender<bool>,
    pub platform: Mutex<Option<Value>>,
    pub privilege: Mutex<Option<Value>>,
    pub sudo_cache: Mutex<Option<(Zeroizing<String>, Instant)>>,
    pub windows_queue: tokio::sync::Mutex<()>,
    pub metrics_state: Mutex<Value>,
    pub stop_latency: Mutex<Option<tokio::task::JoinHandle<()>>>,
}

impl SessionRecord {
    pub fn has_jump(&self) -> bool {
        self.route.len() > 1
    }

    pub fn lost(&self) -> watch::Receiver<bool> {
        self.lost_tx.subscribe()
    }

    pub fn is_connected(&self) -> bool {
        !self.finalized.load(Ordering::SeqCst) && *self.state.lock() == "connected"
    }

    pub fn handle(&self) -> CoreResult<Arc<Handle<ClientHandler>>> {
        self.handle.lock().clone().ok_or_else(not_connected)
    }

    pub fn public_status(&self) -> Value {
        json!({
            "sessionId": self.session_id,
            "profileId": self.profile_id,
            "label": self.label,
            "host": self.host,
            "port": self.port,
            "username": self.username,
            "authMethod": self.auth_method,
            "state": *self.state.lock(),
            "slot": self.slot,
            "adhoc": self.adhoc,
            "jump": self.route.iter().take(self.route.len().saturating_sub(1)).map(|p| json!({
                "label": p["label"], "host": p["host"], "port": p["port"], "username": p["username"]
            })).collect::<Vec<_>>(),
            "latency": self.latency.lock().clone().unwrap_or_else(|| json!({ "pingMs": null, "sshRttMs": null, "measuredAt": null }))
        })
    }
}

pub fn not_connected() -> CoreError {
    CoreError::new("SESSION_NOT_CONNECTED", "The terminal session is not connected.")
}

#[derive(Clone)]
struct PendingTrust {
    profile_id: String,
    host: String,
    port: u16,
    fingerprint: String,
    known: Option<String>,
    expires: Instant,
}

pub type Emitter = Arc<dyn Fn(&str, Value) + Send + Sync>;

#[derive(Clone)]
enum Credentials {
    Password(Zeroizing<String>),
    Key(Arc<russh::keys::PrivateKey>),
}

// The lease closes every opened transport on errors, cancellation and normal teardown.
// Reverse order always disconnects the target before the jumps.
#[derive(Default)]
struct OpenedRoute {
    handles: Vec<Arc<Handle<ClientHandler>>>,
    address: String,
}

impl Drop for OpenedRoute {
    fn drop(&mut self) {
        let handles = std::mem::take(&mut self.handles);
        if !handles.is_empty() {
            // No runtime exists during process shutdown; the OS then closes the sockets.
            let Ok(runtime) = tokio::runtime::Handle::try_current() else { return };
            runtime.spawn(async move {
                for handle in handles.into_iter().rev() {
                    let _ = tokio::time::timeout(Duration::from_secs(2), handle.disconnect(russh::Disconnect::ByApplication, "", "en")).await;
                }
            });
        }
    }
}

/// Internal probe snapshot. Never serialized or exposed over IPC. A Windows sshd
/// restart may close the original session before its guarded login probe runs.
pub struct ConnectionTarget {
    route: Vec<Value>,
    jump_credentials: Vec<Credentials>,
    address: String,
}

pub struct SessionManager {
    pub profiles: Arc<ProfileService>,
    pub logger: Arc<Logger>,
    emit: Emitter,
    sessions: Mutex<HashMap<String, Arc<SessionRecord>>>,
    pending_trust: Mutex<HashMap<String, PendingTrust>>,
    /// Internal subscribers (Transfer Manager) receive every public state change.
    pub state_events: broadcast::Sender<Value>,
}

impl SessionManager {
    pub fn new(profiles: Arc<ProfileService>, logger: Arc<Logger>, emit: Emitter) -> Arc<Self> {
        let (state_events, _) = broadcast::channel(256);
        Arc::new(Self { profiles, logger, emit, sessions: Mutex::new(HashMap::new()), pending_trust: Mutex::new(HashMap::new()), state_events })
    }

    fn emit_state(&self, status: Value) {
        let _ = self.state_events.send(status.clone());
        (self.emit)("ssh:state", status);
    }

    /// Re-publishes the state of a live session (latency updates).
    pub fn publish_state(&self, record: &Arc<SessionRecord>) {
        let current = self.sessions.lock().get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false);
        if current && !record.finalized.load(Ordering::SeqCst) {
            self.emit_state(record.public_status());
        }
    }

    pub fn list(&self) -> Vec<Value> {
        self.sessions.lock().values().map(|r| r.public_status()).collect()
    }

    pub fn get(&self, session_id: &str) -> Option<Arc<SessionRecord>> {
        self.sessions.lock().get(session_id).cloned()
    }

    pub fn require_connected(&self, session_id: &str) -> CoreResult<Arc<SessionRecord>> {
        let record = self.get(session_id).ok_or_else(not_connected)?;
        if !record.is_connected() || record.shell.lock().is_none() {
            return Err(not_connected());
        }
        Ok(record)
    }

    fn credentials(&self, profile: &Value, options: &Value) -> CoreResult<Credentials> {
        self.credentials_fields(profile, options.get("password"), options.get("passphrase"))
    }

    fn credentials_fields(&self, profile: &Value, password: Option<&Value>, passphrase: Option<&Value>) -> CoreResult<Credentials> {
        if profile["authMethod"] == "privateKey" {
            let path = profile["privateKeyPath"].as_str().unwrap_or("");
            let text = Zeroizing::new(read_private_key_file(path)?);
            let passphrase = passphrase.and_then(Value::as_str).map(|s| Zeroizing::new(s.to_string()));
            let key = decode_private_key(&text, passphrase.as_deref().map(|s| s.as_str()))?;
            Ok(Credentials::Key(Arc::new(key)))
        } else {
            let id = profile["id"].as_str().unwrap_or("");
            let password = password
                .and_then(Value::as_str)
                .filter(|p| !p.is_empty())
                .map(str::to_string)
                .or_else(|| if id.starts_with("adhoc:") { None } else { self.profiles.get_stored_password(id).ok().flatten() })
                .unwrap_or_default();
            if password.is_empty() {
                return Err(CoreError::new("PASSWORD_REQUIRED", "Enter the SSH password."));
            }
            Ok(Credentials::Password(Zeroizing::new(password)))
        }
    }

    async fn resolve_address(&self, host: &str, port: u16) -> CoreResult<IpAddr> {
        if let Ok(ip) = host.parse::<IpAddr>() {
            return Ok(ip);
        }
        let addresses: Vec<SocketAddr> = match tokio::net::lookup_host((host, port)).await {
            Ok(list) => list.collect(),
            Err(_) => Vec::new(),
        };
        let mut ips: Vec<IpAddr> = addresses.iter().map(|a| a.ip()).collect();
        let mut source = "system";
        if ips.is_empty() {
            ips = crate::core::diagnostics::dns::resolve_both(host).await;
            source = "dns";
        }
        let selected = ips.iter().find(|ip| ip.is_ipv4()).or_else(|| ips.first()).copied().ok_or_else(|| {
            CoreError::new("SSH_DNS_RESOLUTION_FAILED", format!("DNS name \"{host}\" could not be resolved.")).with("host", host)
        })?;
        self.logger.info(
            "SSH hostname resolved.",
            json!({ "host": host, "address": selected.to_string(), "addressFamily": if selected.is_ipv4() { 4 } else { 6 }, "addressCount": ips.len(), "source": source }),
        );
        Ok(selected)
    }

    fn normalize_error(&self, error: &russh::Error, details: &[(&str, Value)]) -> CoreError {
        let message = error.to_string();
        let mut code = "SSH_CONNECTION_FAILED";
        if let russh::Error::IO(io) = error {
            let raw = io.raw_os_error();
            if cfg!(target_os = "macos") && matches!(raw, Some(1) | Some(13)) {
                code = "SSH_LOCAL_NETWORK_DENIED";
            } else if cfg!(target_os = "macos") && matches!(raw, Some(65) | Some(51)) {
                code = "SSH_MAC_NETWORK_UNREACHABLE";
            } else if io.kind() == std::io::ErrorKind::ConnectionRefused {
                code = "SSH_CONNECTION_REFUSED";
            } else if io.kind() == std::io::ErrorKind::TimedOut {
                code = "SSH_TIMEOUT";
            }
        } else if matches!(error, russh::Error::ConnectionTimeout | russh::Error::KeepaliveTimeout | russh::Error::InactivityTimeout | russh::Error::Elapsed(_)) {
            code = "SSH_TIMEOUT";
        } else if matches!(error, russh::Error::NotAuthenticated | russh::Error::NoAuthMethod) {
            code = "SSH_AUTHENTICATION_FAILED";
        }
        let mut result = CoreError::new(code, message);
        for (key, value) in details {
            result = result.with(key, value.clone());
        }
        result
    }

    fn manual_profile(&self, input: &Value) -> CoreResult<Value> {
        if !input.is_object() || !matches!(input["authMethod"].as_str(), Some("password" | "privateKey")) {
            return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid manual SSH target."));
        }
        let mut profile = normalize_profile(input, None)
            .map_err(|e| CoreError::new("SSH_TARGET_INVALID", e.message))?;
        profile["id"] = json!(format!("adhoc:{}", uuid::Uuid::new_v4()));
        profile["label"] = json!(format!("{}@{}", profile["username"].as_str().unwrap_or(""), profile["host"].as_str().unwrap_or("")));
        profile["jumpProfileId"] = json!("");
        Ok(profile)
    }

    fn route_profiles(&self, options: &Value) -> CoreResult<Vec<Value>> {
        let profile = match (options.get("profileId"), options.get("adhoc")) {
            (Some(Value::String(id)), None) if !id.is_empty() => self.profiles.get(id)?,
            (None, Some(input)) => self.manual_profile(input)?,
            _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Choose exactly one saved or manual SSH target.")),
        };
        let mut seen = std::collections::HashSet::from([profile["id"].as_str().unwrap_or("").to_string()]);
        let saved_jump = |id: &str| self.profiles.get(id).map_err(|_| CoreError::new("JUMP_PROFILE_NOT_FOUND", "Jump profile was not found.")
            .with("hop", "jump").with("hopHost", "").with("jumpProfileId", id));
        let mut next = match options.get("jump") {
            Some(Value::Null) => None,
            None => profile["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?,
            Some(jump) if jump.is_object() && jump.get("profileId").is_some() => {
                if jump.get("host").is_some() {
                    return Err(CoreError::new("SSH_TARGET_INVALID", "Choose one jump host."));
                }
                Some(saved_jump(jump["profileId"].as_str().unwrap_or(""))?)
            }
            Some(jump) => Some(self.manual_profile(jump)?),
        };
        let mut jumps = Vec::new();
        while let Some(hop) = next {
            if !seen.insert(hop["id"].as_str().unwrap_or("").to_string()) {
                return Err(self.hop_error(CoreError::new("JUMP_LOOP", "The jump route contains a cycle."), &hop, true));
            }
            if jumps.len() == 3 {
                return Err(self.hop_error(CoreError::new("JUMP_TOO_DEEP", "At most three jump hosts are allowed."), &hop, true));
            }
            next = hop["jumpProfileId"].as_str().filter(|id| !id.is_empty()).map(saved_jump).transpose()?;
            jumps.push(hop);
        }
        jumps.reverse();
        jumps.push(profile);
        Ok(jumps)
    }

    fn hop_error(&self, error: CoreError, profile: &Value, jump: bool) -> CoreError {
        let mut error = error.with("host", profile["host"].clone()).with("port", profile["port"].clone())
            .with("hop", if jump { "jump" } else { "target" });
        if jump {
            error = error.with("hopHost", profile["host"].clone());
        }
        error
    }

    fn route_credentials(&self, route: &[Value], options: &Value) -> CoreResult<Vec<Credentials>> {
        route[..route.len() - 1].iter().enumerate().map(|(i, hop)| {
            let (password, passphrase) = if i + 2 == route.len() {
                (options.get("jumpPassword"), options.get("jumpPassphrase"))
            } else { (None, None) };
            self.credentials_fields(hop, password, passphrase).map_err(|e| self.hop_error(e, hop, true))
        }).collect()
    }

    /// Shared handshake, host-key verification and authentication for terminals and
    /// fresh logins. Only the first transport is resolved locally.
    async fn open_route(&self, route: &[Value], jumps: &[Credentials], target: &Credentials, address: Option<&str>) -> CoreResult<OpenedRoute> {
        let mut opened = OpenedRoute::default();
        for (i, profile) in route.iter().enumerate() {
            let jump = i + 1 < route.len();
            let host = profile["host"].as_str().unwrap_or("");
            let port = profile["port"].as_u64().unwrap_or(22) as u16;
            let known = self.profiles.get_host_key(host, port).map_err(|e| self.hop_error(e, profile, jump))?;
            let observed = Arc::new(Mutex::new(None));
            let handler = ClientHandler { expected: known.clone(), observed: observed.clone() };
            let creds = if jump { &jumps[i] } else { target };
            let attempt = tokio::time::timeout(READY_TIMEOUT, async {
                let mut handle = if let Some(previous) = opened.handles.last() {
                    let channel = previous.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0).await
                        .map_err(|_| ConnectFailure::Forward)?;
                    client::connect_stream(client_config(true), channel.into_stream(), handler).await.map_err(ConnectFailure::Ssh)?
                } else {
                    let ip = match address {
                        Some(address) => address.parse::<IpAddr>().map_err(|_| ConnectFailure::Core(CoreError::new("SSH_CONNECTION_FAILED", "Invalid first-hop address.")))?,
                        None => self.resolve_address(host, port).await.map_err(ConnectFailure::Core)?,
                    };
                    opened.address = ip.to_string();
                    client::connect(client_config(true), SocketAddr::new(ip, port), handler).await.map_err(ConnectFailure::Ssh)?
                };
                let result = match creds {
                    Credentials::Password(password) => handle.authenticate_password(profile["username"].as_str().unwrap_or(""), password.as_str()).await,
                    Credentials::Key(key) => {
                        let hash = handle.best_supported_rsa_hash().await.map_err(ConnectFailure::Ssh)?.flatten();
                        handle.authenticate_publickey(profile["username"].as_str().unwrap_or(""), PrivateKeyWithHashAlg::new(key.clone(), hash)).await
                    }
                }.map_err(ConnectFailure::Ssh)?;
                if !result.success() {
                    let _ = handle.disconnect(russh::Disconnect::ByApplication, "", "en").await;
                    return Err(ConnectFailure::Auth);
                }
                Ok::<_, ConnectFailure>(handle)
            }).await;
            match attempt {
                Ok(Ok(handle)) => opened.handles.push(Arc::new(handle)),
                failure => {
                    let fingerprint = observed.lock().clone();
                    let error = match failure {
                        Ok(Err(ConnectFailure::Ssh(russh::Error::UnknownKey))) if fingerprint.is_some() => {
                            let mut error = CoreError::new(if known.is_some() { "HOST_KEY_MISMATCH" } else { "HOST_KEY_UNKNOWN" }, "SSH host key verification requires confirmation.")
                                .with("fingerprint", fingerprint.unwrap());
                            if let Some(known) = known { error = error.with("knownFingerprint", known); }
                            error
                        }
                        Ok(Err(ConnectFailure::Forward)) => {
                            // The refusal belongs to the forwarding jump, not the next host.
                            return Err(self.hop_error(CoreError::new("JUMP_CHANNEL_FAILED", "The jump host refused forwarding or could not reach the next host.").with("nextHost", host).with("nextPort", port), &route[i - 1], true));
                        }
                        Ok(Err(ConnectFailure::Ssh(e))) => self.normalize_error(&e, &[]),
                        Ok(Err(ConnectFailure::Auth)) => CoreError::new("SSH_AUTHENTICATION_FAILED", "All configured authentication methods failed."),
                        Ok(Err(ConnectFailure::Core(e))) => e,
                        Err(_) => CoreError::new("SSH_TIMEOUT", "Timed out while waiting for handshake."),
                        Ok(Ok(_)) => unreachable!(),
                    };
                    return Err(self.hop_error(error, profile, jump));
                }
            }
        }
        Ok(opened)
    }

    pub async fn connect(self: &Arc<Self>, options: &Value) -> CoreResult<Value> {
        let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").trim().to_string();
        if !SESSION_ID.is_match(&session_id) {
            return Err(CoreError::new("INVALID_SESSION_ID", "Invalid terminal session ID."));
        }
        let slot = match options.get("slot") {
            None | Some(Value::Null) => None,
            Some(v) if v.as_u64() == Some(1) || v.as_u64() == Some(2) => Some(v.as_u64().unwrap() as u8),
            _ => return Err(CoreError::new("SSH_TARGET_INVALID", "Invalid terminal slot.")),
        };
        let route = self.route_profiles(options)?;
        let profile = route.last().unwrap().clone();
        let jumps = self.route_credentials(&route, options)?;
        let credentials = self.credentials(&profile, options).map_err(|e| self.hop_error(e, &profile, false))?;
        let (lost_tx, _) = watch::channel(false);
        let record = Arc::new(SessionRecord {
            session_id: session_id.clone(),
            profile_id: profile["id"].as_str().unwrap_or("").to_string(),
            label: profile["label"].as_str().unwrap_or("").to_string(),
            host: profile["host"].as_str().unwrap_or("").to_string(),
            address: Mutex::new(profile["host"].as_str().unwrap_or("").to_string()),
            port: profile["port"].as_u64().unwrap_or(22) as u16,
            username: profile["username"].as_str().unwrap_or("").to_string(),
            auth_method: profile["authMethod"].as_str().unwrap_or("password").to_string(),
            slot,
            adhoc: profile["id"].as_str().unwrap_or("").starts_with("adhoc:"),
            profile,
            route,
            jump_credentials: Mutex::new(None),
            transport: Mutex::new(None),
            state: Mutex::new("connecting".into()),
            handle: Mutex::new(None),
            shell: Mutex::new(None),
            latency: Mutex::new(None),
            finalized: AtomicBool::new(false),
            lifecycle: Mutex::new(()),
            lost_tx,
            platform: Mutex::new(None),
            privilege: Mutex::new(None),
            sudo_cache: Mutex::new(None),
            windows_queue: tokio::sync::Mutex::new(()),
            metrics_state: Mutex::new(json!({})),
            stop_latency: Mutex::new(None),
        });
        let previous = {
            let mut sessions = self.sessions.lock();
            self.pending_trust.lock().remove(&session_id);
            sessions.insert(session_id.clone(), record.clone())
        };
        if let Some(previous) = previous { self.finalize(&previous, "reconnect"); }
        self.emit_state(record.public_status());
        let cols = normalize_terminal_size(options.get("cols"), 100, 20, 500);
        let rows = normalize_terminal_size(options.get("rows"), 30, 5, 300);
        let mut lost = record.lost();
        let result = {
            let attempt = async {
                let transport = self.open_route(&record.route, &jumps, &credentials, None).await?;
                let handle = transport.handles.last().unwrap().clone();
                let channel = tokio::time::timeout(READY_TIMEOUT, async {
                    let channel = handle.channel_open_session().await?;
                    channel.request_pty(false, "xterm-256color", cols, rows, 0, 0, &[]).await?;
                    channel.request_shell(true).await?;
                    Ok::<_, russh::Error>(channel)
                }).await.map_err(|_| CoreError::new("SSH_TIMEOUT", "Timed out opening the terminal."))?
                    .map_err(|e| self.normalize_error(&e, &[]))?;
                Ok::<_, CoreError>((transport, handle, channel))
            };
            let cancelled = *lost.borrow();
            if cancelled { Err(not_connected()) } else {
                tokio::select! {
                    result = attempt => result,
                    _ = lost.changed() => Err(not_connected()),
                }
            }
        };
        match result {
            Ok((transport, handle, channel)) => {
                let status = {
                    let _lifecycle = record.lifecycle.lock();
                    let sessions = self.sessions.lock();
                    let current = sessions.get(&session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
                    if !current || record.finalized.load(Ordering::SeqCst) { return Err(not_connected()); }
                    self.pending_trust.lock().remove(&session_id);
                    drop(sessions);
                    if !record.has_jump() { *record.address.lock() = transport.address.clone(); }
                    *record.handle.lock() = Some(handle);
                    *record.transport.lock() = Some(transport);
                    *record.jump_credentials.lock() = Some(jumps);
                    *record.state.lock() = "connected".into();
                    self.bind_shell(&record, channel);
                    if !record.adhoc { let _ = self.profiles.mark_used(&record.profile_id); }
                    record.public_status()
                };
                self.logger.info("SSH session connected.", json!({ "sessionId": record.session_id, "profileId": record.profile_id, "host": record.host, "port": record.port, "username": record.username, "authMethod": record.auth_method }));
                self.publish_state(&record);
                crate::core::ssh::latency::start(self.clone(), record.clone());
                let manager = self.clone();
                tokio::spawn(async move {
                    let mut lost = record.lost();
                    loop {
                        tokio::select! {
                            _ = lost.changed() => break,
                            _ = tokio::time::sleep(Duration::from_millis(250)) => {
                                let closed = record.transport.lock().as_ref().map(|t| t.handles.iter().any(|h| h.is_closed())).unwrap_or(true);
                                if closed { manager.finalize(&record, "route-close"); break; }
                            }
                        }
                    }
                });
                Ok(status)
            }
            Err(error) => {
                let error = error.with("sessionId", session_id.clone()).with("profileId", if record.adhoc { "" } else { &record.profile_id });
                let lifecycle = record.lifecycle.lock();
                let sessions = self.sessions.lock();
                let current = sessions.get(&record.session_id).is_some_and(|r| Arc::ptr_eq(r, &record));
                if current && !record.finalized.load(Ordering::SeqCst) && matches!(error.code.as_str(), "HOST_KEY_UNKNOWN" | "HOST_KEY_MISMATCH") {
                    self.pending_trust.lock().insert(session_id, PendingTrust {
                        profile_id: if record.adhoc { String::new() } else { record.profile_id.clone() },
                        host: error.details["host"].as_str().unwrap_or("").to_string(),
                        port: error.details["port"].as_u64().unwrap_or(22) as u16,
                        fingerprint: error.details["fingerprint"].as_str().unwrap_or("").to_string(),
                        known: error.details.get("knownFingerprint").and_then(Value::as_str).map(str::to_string),
                        expires: Instant::now() + PENDING_TRUST_TTL,
                    });
                }
                drop(sessions);
                drop(lifecycle);
                self.logger.warn("SSH connection failed.", json!({ "sessionId": record.session_id, "code": error.code, "message": error.message, "hop": error.details.get("hop"), "host": error.details.get("host"), "port": error.details.get("port") }));
                self.finalize(&record, &error.code);
                Err(error)
            }
        }
    }

    fn bind_shell(self: &Arc<Self>, record: &Arc<SessionRecord>, channel: russh::Channel<Msg>) {
        let (tx, mut rx) = mpsc::unbounded_channel::<ShellCommand>();
        *record.shell.lock() = Some(tx);
        let (mut read, write): (_, ChannelWriteHalf<Msg>) = channel.split();
        let manager = self.clone();
        let record = record.clone();
        tokio::spawn(async move {
            let mut decoder = Utf8Decoder::default();
            let mut lost = record.lost();
            let reason = loop {
                if !record.is_connected() { break "disconnected"; }
                tokio::select! {
                    message = read.wait() => match message {
                        Some(ChannelMsg::Data { data }) => {
                            let text = decoder.write(&data);
                            if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
                        }
                        Some(ChannelMsg::ExtendedData { data, .. }) => {
                            let text = decoder.write(&data);
                            if !text.is_empty() && record.is_connected() { (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": text })); }
                        }
                        Some(ChannelMsg::Close) | None => break "shell-close",
                        Some(ChannelMsg::Failure) => {
                            (manager.emit)("ssh:error", json!({ "sessionId": record.session_id, "message": "The SSH server refused the interactive shell." }));
                            break "shell-error";
                        }
                        Some(_) => {}
                    },
                    command = rx.recv() => match command {
                        Some(ShellCommand::Input(bytes)) => { if write.data_bytes(bytes).await.is_err() { break "shell-error"; } }
                        Some(ShellCommand::Resize(cols, rows)) => { let _ = write.window_change(cols, rows, 0, 0).await; }
                        None => break "shell-close",
                    },
                    _ = lost.changed() => break "disconnected",
                }
            };
            let trailing = decoder.end();
            if !trailing.is_empty() && record.is_connected() {
                (manager.emit)("ssh:data", json!({ "sessionId": record.session_id, "data": trailing }));
            }
            let _ = write.close().await;
            manager.finalize(&record, reason);
        });
    }

    pub fn trust_pending(&self, options: &Value) -> CoreResult<Value> {
        let session_id = options.get("sessionId").and_then(Value::as_str).unwrap_or("").to_string();
        let fingerprint = options.get("fingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
        let pending = self.pending_trust.lock().get(&session_id).cloned();
        let Some(pending) = pending.filter(|p| p.expires >= Instant::now()) else {
            self.pending_trust.lock().remove(&session_id);
            return Err(CoreError::new("HOST_KEY_CONFIRMATION_EXPIRED", "The host-key confirmation expired."));
        };
        if options.get("profileId").and_then(Value::as_str) != Some(pending.profile_id.as_str()) || pending.fingerprint != fingerprint {
            return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "The host-key confirmation does not match the pending connection."));
        }
        let replace = options.get("replaceKnown") == Some(&Value::Bool(true));
        if let Some(known) = &pending.known {
            let confirmed = options.get("knownFingerprint").and_then(Value::as_str).unwrap_or("").to_lowercase();
            if !replace || &confirmed != known {
                return Err(CoreError::new(
                    "HOST_KEY_REPLACEMENT_CONFIRMATION_REQUIRED",
                    "Replacing a changed SSH host key requires explicit confirmation of both fingerprints.",
                ));
            }
            if self.profiles.get_host_key(&pending.host, pending.port)?.as_deref() != Some(known.as_str()) {
                self.pending_trust.lock().remove(&session_id);
                return Err(CoreError::new("HOST_KEY_REPLACEMENT_STALE", "The stored SSH host key changed while confirmation was pending."));
            }
        } else if replace {
            return Err(CoreError::new("HOST_KEY_CONFIRMATION_INVALID", "A new host key cannot be confirmed as a replacement."));
        }
        let result = self.profiles.trust_host_key(&pending.host, pending.port, &pending.fingerprint)?;
        self.pending_trust.lock().remove(&session_id);
        let details = json!({ "host": pending.host, "port": pending.port, "fingerprint": pending.fingerprint, "replacedFingerprint": pending.known });
        if pending.known.is_some() {
            self.logger.warn("Changed SSH host key replaced by user confirmation.", details);
        } else {
            self.logger.info("SSH host key trusted by user.", details);
        }
        Ok(result)
    }

    /// Terminal input. Called synchronously and in order; data is queued to the shell task.
    pub fn write(&self, session_id: &str, data: &str) -> CoreResult<()> {
        let record = self.require_connected(session_id)?;
        if data.len() > MAX_INPUT_LENGTH {
            return Err(CoreError::new("TERMINAL_INPUT_TOO_LARGE", "Terminal input is too large."));
        }
        let sender = record.shell.lock().clone().ok_or_else(not_connected)?;
        sender.send(ShellCommand::Input(data.as_bytes().to_vec())).map_err(|_| not_connected())
    }

    pub fn resize(&self, session_id: &str, cols: Option<&Value>, rows: Option<&Value>) -> CoreResult<Value> {
        let record = self.require_connected(session_id)?;
        let cols = normalize_terminal_size(cols, 100, 20, 500);
        let rows = normalize_terminal_size(rows, 30, 5, 300);
        if let Some(sender) = record.shell.lock().clone() {
            let _ = sender.send(ShellCommand::Resize(cols, rows));
        }
        Ok(json!({ "cols": cols, "rows": rows }))
    }

    /// The fingerprint this session's server is trusted with now.
    pub fn trusted_host_key(&self, session_id: &str) -> Option<String> {
        let record = self.get(session_id)?;
        self.profiles.get_host_key(&record.host, record.port).ok().flatten()
    }

    /// Availability of a jump-only server, through a session already open to its jump host:
    /// one forwarding channel, no login and no new session. Never an unproven "online".
    pub async fn probe_through_jump(&self, jump_profile_id: &str, host: &str, port: u16, timeout: Duration) -> Outcome {
        let handle = self.sessions.lock().values()
            .find(|record| record.profile_id == jump_profile_id && record.state.lock().as_str() == "connected")
            .and_then(|record| record.handle.lock().clone());
        let Some(handle) = handle else { return Outcome::Unverified { reason: "noJumpSession" } };
        let start = std::time::Instant::now();
        match tokio::time::timeout(timeout, handle.channel_open_direct_tcpip(host, u32::from(port), "127.0.0.1", 0)).await {
            Ok(Ok(channel)) => {
                let ms = (start.elapsed().as_secs_f64() * 10_000.0).round() / 10.0;
                let _ = channel.close().await;
                Outcome::Online { ms }
            }
            Ok(Err(russh::Error::ChannelOpenFailure(russh::ChannelOpenFailure::ConnectFailed))) => Outcome::Failed { reason: "refused" },
            Ok(Err(_)) => Outcome::Unverified { reason: "jumpDenied" },
            Err(_) => Outcome::Failed { reason: "timeout" },
        }
    }

    pub async fn disconnect(&self, session_id: &str, reason: &str) -> Value {
        self.pending_trust.lock().remove(session_id);
        let record = self.sessions.lock().get(session_id).cloned();
        match record {
            Some(record) => {
                self.finalize(&record, reason);
                json!({ "disconnected": true })
            }
            None => json!({ "disconnected": false }),
        }
    }

    pub async fn disconnect_all(&self, reason: &str) -> Value {
        let ids: Vec<String> = self.sessions.lock().keys().cloned().collect();
        for id in &ids {
            self.disconnect(id, reason).await;
        }
        self.pending_trust.lock().clear();
        json!({ "disconnected": ids.len() })
    }

    pub fn finalize(&self, record: &Arc<SessionRecord>, reason: &str) {
        let lifecycle = record.lifecycle.lock();
        if record.finalized.swap(true, Ordering::SeqCst) {
            return;
        }
        *record.state.lock() = "disconnected".into();
        if let Some(task) = record.stop_latency.lock().take() {
            task.abort();
        }
        *record.latency.lock() = None;
        record.lost_tx.send_replace(true);
        *record.sudo_cache.lock() = None;
        {
            let mut sessions = self.sessions.lock();
            if sessions.get(&record.session_id).map(|r| Arc::ptr_eq(r, record)).unwrap_or(false) {
                sessions.remove(&record.session_id);
            }
        }
        record.shell.lock().take();
        record.handle.lock().take();
        record.transport.lock().take();
        record.jump_credentials.lock().take();
        let mut status = record.public_status();
        status["reason"] = Value::String(reason.to_string());
        drop(lifecycle);
        self.emit_state(status);
        self.logger.info("SSH session disconnected.", json!({ "sessionId": record.session_id, "profileId": record.profile_id, "reason": reason }));
    }

    // -----------------------------------------------------------------------------------------
    // Exec channels
    // -----------------------------------------------------------------------------------------

    pub async fn exec_channel(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecChannel> {
        let handle = record.handle()?;
        let lost = record.lost();
        let opened = tokio::time::timeout(timeout, async {
            let channel = handle.channel_open_session().await?;
            channel.exec(true, command.as_bytes().to_vec()).await?;
            Ok::<_, russh::Error>(channel)
        })
        .await
        .map_err(|_| CoreError::new("REMOTE_COMMAND_TIMEOUT", "The remote command timed out."))?
        .map_err(|e| if record.is_connected() { CoreError::new("REMOTE_COMMAND_FAILED", e.to_string()) } else { not_connected() })?;
        let (read, write) = opened.split();
        Ok(ExecChannel::new(read, write, timeout, limit, lost))
    }

    /// Runs a fixed command and requires exit code 0 (`_execFixed`).
    pub async fn exec_fixed(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration, limit: usize) -> CoreResult<ExecResult> {
        let mut channel = self.exec_channel(record, command, timeout, limit).await?;
        channel.eof().await;
        let result = channel.finish().await?;
        match result.code {
            Some(0) => Ok(result),
            code => {
                let message = if result.stderr.trim().is_empty() { format!("Remote command exited with {}.", code.map(|c| c.to_string()).unwrap_or_else(|| "undefined".into())) } else { result.stderr.trim().to_string() };
                Err(CoreError::new("REMOTE_COMMAND_FAILED", message).with("remoteExitCode", code.map(Value::from).unwrap_or(Value::Null)))
            }
        }
    }

    pub async fn exec_fixed_default(&self, record: &Arc<SessionRecord>, command: &str, timeout: Duration) -> CoreResult<ExecResult> {
        self.exec_fixed(record, command, timeout, MAX_COMMAND_OUTPUT).await
    }

    /// One transport round trip (keepalive@openssh.com with want-reply), fallback `exit 0`.
    pub async fn ssh_round_trip(&self, record: &Arc<SessionRecord>) -> CoreResult<(Option<f64>, &'static str)> {
        let handle = record.handle()?;
        let started = Instant::now();
        match tokio::time::timeout(Duration::from_secs(5), handle.send_ping()).await {
            Ok(Ok(())) => return Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "transport")),
            Ok(Err(_)) => {}
            Err(_) => return Err(CoreError::new("SSH_RTT_TIMEOUT", "The SSH server did not answer the round-trip request in time.")),
        }
        let started = Instant::now();
        self.exec_fixed_default(record, "exit 0", Duration::from_secs(5)).await?;
        Ok((Some(started.elapsed().as_secs_f64() * 1000.0), "command"))
    }

    /// Serializes PowerShell helpers of one Windows session (Win32-OpenSSH stalls on concurrent stdin helpers).
    pub async fn windows_serial<T, F: std::future::Future<Output = CoreResult<T>>>(&self, record: &Arc<SessionRecord>, task: F) -> CoreResult<T> {
        let _turn = record.windows_queue.lock().await;
        task.await
    }

    // -----------------------------------------------------------------------------------------
    // Identity facts, fresh logins (Users lockout protection)
    // -----------------------------------------------------------------------------------------

    pub fn session_auth_info(&self, session_id: &str) -> CoreResult<Value> {
        let record = self.require_connected(session_id)?;
        let fingerprint = if record.auth_method == "privateKey" {
            crate::core::ssh::keys::public_key_fingerprint_for_private_key(record.profile["privateKeyPath"].as_str().unwrap_or("")).map(Value::String).unwrap_or(Value::Null)
        } else { Value::Null };
        // The target address is not the SSH client's source address. Users policy
        // obtains the actual source from the remote SSH_CONNECTION environment.
        Ok(json!({ "username": record.username, "authMethod": record.auth_method, "address": record.address.lock().clone(), "publicKeyFingerprint": fingerprint }))
    }

    pub fn probe_login_requirements(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
        let record = self.require_connected(session_id)?;
        Ok(match self.credentials(&record.profile, credentials) {
            Ok(_) => json!({ "ready": true, "missing": null }),
            Err(e) => json!({ "ready": false, "missing": if e.code == "PASSWORD_REQUIRED" { "password" } else if matches!(e.code.as_str(), "PRIVATE_KEY_PASSPHRASE_REQUIRED" | "PRIVATE_KEY_PASSPHRASE_INVALID") { "passphrase" } else { "privateKey" } }),
        })
    }

    pub fn connection_target(&self, session_id: &str) -> CoreResult<ConnectionTarget> {
        let record = self.require_connected(session_id)?;
        let _lifecycle = record.lifecycle.lock();
        if !record.is_connected() { return Err(not_connected()); }
        let jump_credentials = record.jump_credentials.lock().clone().ok_or_else(not_connected)?;
        let address = record.address.lock().clone();
        Ok(ConnectionTarget {
            route: record.route.clone(),
            jump_credentials,
            address,
        })
    }

    /// One fixed command over a new connection following the original verified route.
    pub async fn fresh_exec(&self, target: &ConnectionTarget, credentials: &Value, command: &str, input: &str, timeout: Duration, accept_any_exit: bool) -> CoreResult<Value> {
        let probe_failed = |error: CoreError| {
            let mut wrapped = CoreError::new("SSH_PROBE_FAILED", format!("A new SSH login was not confirmed: {}", error.message));
            wrapped.details = error.details;
            wrapped.with("causeCode", error.code)
        };
        let run = async {
            let profile = target.route.last().ok_or_else(not_connected)?;
            let creds = self.credentials(profile, credentials)?;
            let address = (target.route.len() == 1).then_some(target.address.as_str());
            let transport = self.open_route(&target.route, &target.jump_credentials, &creds, address).await?;
            let handle = transport.handles.last().unwrap();
            let channel = handle.channel_open_session().await.map_err(|e| self.normalize_error(&e, &[]))?;
            channel.exec(true, command.as_bytes().to_vec()).await.map_err(|e| self.normalize_error(&e, &[]))?;
            let (read, write) = channel.split();
            let (_tx, lost) = watch::channel(false);
            let mut exec = ExecChannel::new(read, write, timeout, MAX_COMMAND_OUTPUT, lost);
            if !input.is_empty() { exec.write(input.as_bytes()).await; }
            exec.eof().await;
            let result = exec.finish().await?;
            if !accept_any_exit && !matches!(result.code, Some(0) | None) {
                return Err(CoreError::new("SSH_PROBE_EXIT", format!("Probe command exited with {}.", result.code.unwrap_or(-1))));
            }
            Ok(json!({ "ok": true, "code": result.code, "stdout": result.stdout, "stderr": result.stderr, "address": target.address, "checkedAt": chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true) }))
        };
        tokio::time::timeout(timeout, run).await.map_err(|_| CoreError::new("SSH_TIMEOUT", "Login probe timed out.")).and_then(|r| r).map_err(probe_failed)
    }

    pub async fn probe_login(&self, session_id: &str, credentials: &Value) -> CoreResult<Value> {
        let target = self.connection_target(session_id)?;
        let result = self.fresh_exec(&target, credentials, "true", "", Duration::from_secs(60), true).await?;
        Ok(json!({ "ok": true, "address": result["address"], "checkedAt": result["checkedAt"] }))
    }
}

enum ConnectFailure {
    Ssh(russh::Error),
    Auth,
    Forward,
    Core(CoreError),
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn terminal_size_bounds() {
        assert_eq!(normalize_terminal_size(Some(&json!(10)), 100, 20, 500), 20);
        assert_eq!(normalize_terminal_size(Some(&json!(9999)), 100, 20, 500), 500);
        assert_eq!(normalize_terminal_size(Some(&json!(80.5)), 100, 20, 500), 100);
        assert_eq!(normalize_terminal_size(None, 30, 5, 300), 30);
    }

    #[test]
    fn quoting() {
        assert_eq!(shell_quote("a'b"), "'a'\"'\"'b'");
    }
}

#[cfg(test)]
#[path = "../../../tests/support/ssh_route_unit.rs"]
mod route_tests;