agentmux_srv/server/app_api/

connections.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! The connection RPCs the frontend's connection UI already calls
//! (`conntypeahead.tsx`, `blockframe.tsx`'s `ConnStatusOverlay`), which had no
//! handler until now (P0 of SPEC_REMOTE_TERMINALS_AND_DURABLE_SESSIONS_2026_10_02.md).
//!
//! A connection AgentMux cannot run is refused with a status the pane shows,
//! instead of the pane silently starting a local shell under a remote name:
//! WSL off Windows or for a distro that is not installed (P1), and SSH where
//! there is no `ssh` (P2).

use super::*;
use crate::backend::mps::Broker;
use crate::backend::remote::remotes;
use crate::backend::remote::sessions;
use crate::backend::remote::status::{self, state};
use crate::backend::remote::ConnTarget;

pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
    // Lists. One call each, with the constant spelled out: the RPC contract
    // test resolves every registered command name statically.
    //
    // The picker's "Remote" section: the user's ssh config hosts and any other
    // SSH connection used this session. The typeahead adds `local` and the
    // WSL distros itself.
    //
    // The requests are `serde_json::Value`: these handlers read what they need
    // loosely (a missing field is ""), and `Value` takes anything, so typing them
    // only records the response (docs/reports/REPORT_MIGRATION_WRAPUP_STATUS_2026_09_16.md
    // §5.4c). The frontend stubs keep their precise request types
    // (generated-manifest.json `dataType`).
    engine.register_typed(COMMAND_CONN_LIST, |_data: serde_json::Value, _ctx| async move {
        let hosts = tokio::task::spawn_blocking(crate::backend::remote::ssh_config::hosts)
            .await
            .unwrap_or_default();
        Ok(remote_names(&hosts, &status::all()))
    });
    engine.register_typed(COMMAND_CONN_LIST_AWS, |_data: serde_json::Value, _ctx| async move {
        Ok(Vec::<String>::new())
    });
    // Installed WSL distributions (empty off Windows, or with no WSL).
    engine.register_typed(COMMAND_WSL_LIST, |_data: serde_json::Value, _ctx| async move {
        Ok(crate::backend::remote::wsl::list().await)
    });

    let broker = state.broker.clone();
    engine.register_typed(COMMAND_CONN_ENSURE, move |data: serde_json::Value, _ctx| {
        let broker = broker.clone();
        // `{ connname, logblockid? }`
        async move { ensure(&broker, &str_field(&data, "connname")).await }
    });

    let app = state.clone();
    engine.register_typed(COMMAND_CONN_CONNECT, move |data: serde_json::Value, _ctx| {
        let app = app.clone();
        // `{ host, keywords?, logblockid? }`
        async move { connect(&app, &str_field(&data, "host"), &str_field(&data, "logblockid")).await }
    });

    let broker = state.broker.clone();
    engine.register_typed(COMMAND_CONN_DISCONNECT, move |data: serde_json::Value, _ctx| {
        let broker = broker.clone();
        async move {
                // The connection name, as a bare string.
                let name = data.as_str().unwrap_or_default().to_string();
                match ConnTarget::parse(&name) {
                    // Closes the shared connection: the sessions on it end,
                    // and the next pane logs in again.
                    Ok(ConnTarget::Ssh(_)) => crate::backend::remote::link::disconnect(&name).await,
                    Ok(ConnTarget::Wsl(_)) => {
                        status::set(Some(&broker), &name, state::DISCONNECTED, None);
                    }
                    _ => {}
                }
                Ok(())
        }
    });

    // A host's durable sessions, for Connectors → Remotes' "Sessions…"
    // (spec §7.6). Listing runs one fixed, read-only command over ssh, as the
    // UI's other connection calls do. Ending one is the user's decision,
    // made in the host's own window (`ask_user`): anything holding srv's
    // auth key can send this RPC, an agent included, but none can answer
    // that window.
    let (store, auth_key) = (state.mstore.clone(), state.auth_key.clone());
    engine.register_typed(COMMAND_CONN_SESSIONS, move |data: serde_json::Value, _ctx| {
        let (store, auth_key) = (store.clone(), auth_key.clone());
        async move {
                let (conn, block) = (str_field(&data, "connname"), str_field(&data, "blockid"));
                let ask = (!block.is_empty()).then_some(sessions::AskIn {
                    block_id: &block,
                    auth_key: &auth_key,
                });
                let list = sessions::list(&store, &conn, ask).await?;
                // Connectors → Remotes shows the count without opening ssh itself.
                remotes::note_sessions(&conn, list.len());
                Ok(list)
        }
    });
    let app = state.clone();
    engine.register_typed(COMMAND_CONN_SESSION_END, move |data: serde_json::Value, _ctx| {
        let app = app.clone();
        async move {
                let (conn, id, block) = (
                    str_field(&data, "connname"),
                    str_field(&data, "sessionid"),
                    str_field(&data, "blockid"),
                );
                if block.is_empty() {
                    return Err("ending a session needs the pane asking, to ask the user in".into());
                }
                if !crate::backend::blockcontroller::durable_ssh::valid_session_id(&id) {
                    return Err(format!("{:?} is not a session id", one_line(&id, 80)));
                }
                confirm_session_end(&app, &block, &conn, &id).await?;
                let ask = sessions::AskIn {
                    block_id: &block,
                    auth_key: &app.auth_key,
                };
                let ended = sessions::end(&conn, &id, Some(ask)).await?;
                Ok(ended)
        }
    });

    // Connectors → Remotes (docs/specs/SPEC_REMOTES_PANE_2026_10_05.md §4.6): its list,
    // a remote's settings, and forgetting a recent one.
    let config = state.config_watcher.clone();
    engine.register_typed(COMMAND_REMOTES_LIST, move |_cmd: CommandRemotesListData, _ctx| {
        let config = config.clone();
        async move { Ok(remotes::list(config.get_full_config().connections.clone()).await) }
    });
    let (config, event_bus, broker) = (state.config_watcher.clone(), state.event_bus.clone(), state.broker.clone());
    engine.register_typed(COMMAND_REMOTE_SET_CONFIG, move |cmd: CommandRemoteSetConfigData, _ctx| {
        let (config, event_bus, broker) = (config.clone(), event_bus.clone(), broker.clone());
        async move {
            let name = remote_name(&cmd.connection)?;
            crate::backend::config_watcher_fs::set_connection_values(&config, &event_bus, &name, &cmd.values)?;
            remotes::publish_change(&broker);
            Ok(())
        }
    });
    let broker = state.broker.clone();
    engine.register_typed(COMMAND_REMOTE_FORGET, move |cmd: CommandRemoteForgetData, _ctx| {
        let broker = broker.clone();
        async move {
            let name = remote_name(&cmd.connection)?;
            if remotes::forget_recent(&name).map_err(|e| format!("could not forget {name}: {e}"))? {
                remotes::publish_change(&broker);
            }
            Ok(())
        }
    });
    // Removing the helper ends the host's sessions, so it is the user's
    // decision, made in the approval window, as ending one session is.
    let app = state.clone();
    engine.register_typed(COMMAND_REMOTE_HELPER_REMOVE, move |cmd: CommandRemoteHelperRemoveData, _ctx| {
        let app = app.clone();
        async move { remove_helper(&app, &cmd.connection, &cmd.blockid).await }
    });
    // Adding a host writes the user's ssh config, which every ssh on the
    // machine reads (git's included), so it is the user's decision, made in
    // the approval window: a new alias could otherwise send another tool's
    // ssh somewhere else.
    let app = state.clone();
    engine.register_typed(COMMAND_REMOTE_ADD, move |cmd: CommandRemoteAddData, _ctx| {
        let app = app.clone();
        async move { add_remote(&app, cmd).await }
    });
    // Testing logs in with the user's keys, so only to a host AgentMux
    // already lists (the user's own ssh config, settings or a connection
    // they made), never one an RPC names.
    let app = state.clone();
    engine.register_typed(COMMAND_REMOTE_TEST, move |cmd: CommandRemoteTestData, _ctx| {
        let app = app.clone();
        async move { test_remote(&app, &cmd.connection, &cmd.blockid).await }
    });
    // Revoking only takes access away (the agent is asked again on its next
    // use of the host), so it needs no question.
    let broker = state.broker.clone();
    engine.register_typed(COMMAND_REMOTE_AGENT_REVOKE, move |cmd: CommandRemoteAgentRevokeData, _ctx| {
        let broker = broker.clone();
        async move {
            let name = ssh_host(&cmd.connection, NOT_SSH)?;
            let revoked = crate::backend::remote::agent_access::revoke(&cmd.agent, &name)
                .map_err(|e| format!("could not revoke {}'s access to {name}: {e}", one_line(&cmd.agent, 60)))?;
            if revoked {
                tracing::info!(agent = %one_line(&cmd.agent, 60), connection = %name, "agent ssh access: revoked");
                remotes::publish_change(&broker);
            }
            Ok(())
        }
    });
    engine.register_typed(COMMAND_REMOTE_SSH_LOCATE, move |cmd: CommandRemoteSshLocateData, _ctx| async move {
        let found = tokio::task::spawn_blocking(move || {
            let (config, ssh_dir) = crate::backend::remote::ssh_config_edit::user_paths()?;
            crate::backend::remote::ssh_config_edit::locate_in(&config, &ssh_dir, &cmd.connection)
        })
        .await
        .map_err(|e| e.to_string())?;
        Ok(found.map(|(path, line)| RemoteSshLocation {
            path: path.to_string_lossy().into_owned(),
            line: line as u32,
        }))
    });
}

/// Add a host to the user's ssh config (§4.4 of SPEC_REMOTES_PANE_2026_10_05.md),
/// once they confirm the exact block in the window of pane `cmd.blockid`.
async fn add_remote(state: &AppState, cmd: CommandRemoteAddData) -> Result<(), String> {
    use crate::backend::remote::ssh_config_edit::{self, NewHost};
    if cmd.blockid.is_empty() {
        return Err("adding a remote needs the pane asking, to ask the user in".into());
    }
    let host = NewHost {
        alias: cmd.alias,
        hostname: cmd.hostname,
        user: cmd.user,
        port: cmd.port,
        identity_file: cmd.identityfile,
        proxy_jump: cmd.proxyjump,
    };
    let block = ssh_config_edit::host_block(&host)?;
    let (config, ssh_dir) =
        ssh_config_edit::user_paths().ok_or_else(|| "there is no home folder for an ssh config".to_string())?;
    let question = serde_json::json!({
        "kind": "consent",
        "title": format!("Add {} to your ssh config", one_line(&host.alias, 80)),
        "message": format!(
            "Add this to the end of {}?\n\n{}\nEvery ssh on this computer will use it.",
            config.display(),
            block
        ),
        "ok_label": "Add",
        "cancel_label": "Cancel",
    });
    let answer = ask_user(state, &cmd.blockid, question).await?;
    if !(answer.answered && answer.approve) {
        return Err("kept: the user chose not to add it".into());
    }
    tokio::task::spawn_blocking(move || ssh_config_edit::add_host_in(&config, &ssh_dir, &host))
        .await
        .map_err(|e| e.to_string())??;
    remotes::publish_change(&state.broker);
    Ok(())
}

const NOT_SSH: &str = "is not an SSH host";

/// `connection`'s canonical name if it is an SSH host; otherwise
/// "<connection> `otherwise`".
fn ssh_host(connection: &str, otherwise: &str) -> Result<String, String> {
    match ConnTarget::parse(connection)? {
        t @ ConnTarget::Ssh(_) => Ok(t.name()),
        _ => Err(format!("{} {otherwise}", one_line(connection, 80))),
    }
}

/// [`ssh_host`] for a host AgentMux already lists (the user's ssh config,
/// settings, or a connection they made), for what logs in with the user's
/// keys: never to one an RPC names, since anything holding srv's key can send
/// one, an agent included.
async fn known_ssh_host(state: &AppState, connection: &str) -> Result<String, String> {
    let name = ssh_host(connection, NOT_SSH)?;
    let known = remotes::list(state.config_watcher.get_full_config().connections.clone())
        .await
        .iter()
        .any(|r| r.kind == "ssh" && r.name == name);
    if !known {
        return Err(format!("{} isn't one of your remotes", one_line(&name, 80)));
    }
    Ok(name)
}

/// Log in to SSH host `name` (a [`known_ssh_host`]) and run `true`, with
/// ssh's prompts going to the user in the window of pane `block`, if any.
/// The connection's status follows (`link::login`), and the connection stays
/// up a while after (`ControlPersist`), so the next pane on it asks nothing.
async fn log_in(state: &AppState, name: &str, block: &str) -> Result<(), String> {
    use crate::backend::remote::{host::HostSsh, link};
    let mut host = HostSsh::for_connection(name)?;
    let _grant = (!block.is_empty()).then(|| host.ask_user_in(block, name, &state.auth_key)).flatten();
    link::login(name, async {
        host.run("true", None, SSH_WITH_PROMPTS).await.and_then(|o| o.ok()).map(|_| ())
    })
    .await
}

/// What `connconnect` does (a Remotes row's Connect, a pane's Reconnect): log
/// in to an SSH host; for anything else, what `connensure` does.
async fn connect(state: &AppState, connection: &str, block: &str) -> Result<(), String> {
    ensure(&state.broker, connection).await?;
    if !matches!(ConnTarget::parse(connection)?, ConnTarget::Ssh(_)) {
        return Ok(());
    }
    let name = known_ssh_host(state, connection).await?;
    log_in(state, &name, block).await
}

/// Log in to `connection` and run `true` (§4.4): success, or ssh's message.
async fn test_remote(state: &AppState, connection: &str, block: &str) -> Result<RemoteTestResult, String> {
    let name = known_ssh_host(state, connection).await?;
    Ok(match log_in(state, &name, block).await {
        Ok(()) => RemoteTestResult { ok: true, message: String::new() },
        Err(message) => RemoteTestResult { ok: false, message },
    })
}

/// Remove AgentMux's helper from SSH host `connection` (Remotes, §4.9 of
/// SPEC_REMOTES_PANE_2026_10_05.md), once the user confirms in the window of
/// pane `block`, naming the sessions that will end.
async fn remove_helper(state: &AppState, connection: &str, block: &str) -> Result<(), String> {
    use crate::backend::remote::{files, helper_hosts, helper_install, host::HostSsh};
    let name = ssh_host(connection, "has no AgentMux helper to remove")?;
    if block.is_empty() {
        return Err("removing the helper needs the pane asking, to ask the user in".into());
    }
    let ask = sessions::AskIn { block_id: block, auth_key: &state.auth_key };
    let running = sessions::list(&state.mstore, &name, Some(ask)).await?;
    let ends = match running.len() {
        0 => "No sessions are running there.".to_string(),
        1 => "The 1 session running there ends, with everything running in it.".to_string(),
        n => format!("The {n} sessions running there end, with everything running in them."),
    };
    let question = serde_json::json!({
        "kind": "consent",
        "title": format!("Remove AgentMux's helper from {}", one_line(&name, 80)),
        "message": format!(
            "Remove AgentMux's helper from {}?\n\n{ends} It deletes ~/.agentmux-remote there; file browsing and terminals that survive disconnects need it installed again.",
            one_line(&name, 80)
        ),
        "ok_label": "Remove",
        "cancel_label": "Keep It",
        // The window draws a destructive answer in the danger tone, not the
        // loud attention fill.
        "destructive": true,
    });
    let answer = ask_user(state, block, question).await?;
    if !(answer.answered && answer.approve) {
        return Err("kept: the user chose not to remove it".into());
    }
    let mut host = HostSsh::for_connection(&name)?;
    let _grant = host.ask_user_in(block, &name, &state.auth_key);
    let out = host
        .run(&helper_install::remove_command(), None, SSH_WITH_PROMPTS)
        .await?
        .ok()?;
    if out.trim() != "ok" {
        return Err(format!("removing the helper from {name} did not finish ({})", one_line(&out, 200)));
    }
    files::close(&name).await;
    if let Err(e) = helper_hosts::forget_helper(&name) {
        tracing::warn!(connection = %name, error = %e, "could not forget the removed helper");
    }
    // Its next need asks again (spec §4.9), whatever was answered before.
    crate::backend::remote::helper_consent::forget_answers(&name);
    remotes::note_sessions(&name, 0);
    remotes::publish_change(&state.broker);
    tracing::info!(connection = %name, sessions = running.len(), "removed the helper");
    Ok(())
}

/// Give the helper install its question (SPEC_REMOTES_PANE_2026_10_05.md
/// §4.9): the approval window, and `conn:helper` from the settings.
pub fn install_helper_consent(state: &AppState) {
    use crate::backend::remote::helper_consent;
    let app = state.clone();
    let config = state.config_watcher.clone();
    let (writer, event_bus, broker) = (state.config_watcher.clone(), state.event_bus.clone(), state.broker.clone());
    helper_consent::install(helper_consent::Deps {
        ask: Box::new(move |block, question| {
            let app = app.clone();
            Box::pin(async move {
                let a = ask_user(&app, &block, question).await?;
                Ok(helper_consent::Answer { answered: a.answered, approve: a.approve, remember: a.checkbox })
            })
        }),
        settings: Box::new(move |connection| {
            let full = config.get_full_config();
            let host = full
                .connections
                .iter()
                .find(|(name, _)| crate::backend::remote::conn::same_connection(name, connection))
                .map(|(_, k)| k.conn_helper.clone())
                .unwrap_or_default();
            (host, full.settings.conn_helper.clone())
        }),
        remember: Box::new(move |connection, value| {
            let mut values = serde_json::Map::new();
            values.insert("conn:helper".into(), serde_json::json!(value));
            crate::backend::config_watcher_fs::set_connection_values(&writer, &event_bus, connection, &values)?;
            remotes::publish_change(&broker);
            Ok(())
        }),
    });
}

/// A connection the Remotes commands may act on: a WSL distribution or an SSH
/// destination, never `local` or something `ConnTarget` refuses.
fn remote_name(connection: &str) -> Result<String, String> {
    match ConnTarget::parse(connection)? {
        ConnTarget::Local => Err("local is not a remote".to_string()),
        _ => Ok(connection.trim().to_string()),
    }
}

/// The user's yes to ending session `id` on `conn`, asked in the window of
/// pane `block`. `Err` unless they said yes.
async fn confirm_session_end(
    state: &AppState,
    block: &str,
    conn: &str,
    id: &str,
) -> Result<(), String> {
    // `consent`: a plain question of AgentMux's own, not a relayed ssh prompt.
    let question = serde_json::json!({
        "kind": "consent",
        "title": format!("End a session on {}", one_line(conn, 80)),
        "message": format!(
            "End the durable session {id} on {}?\n\nIts shell, and everything running in it, stops. This cannot be undone.",
            one_line(conn, 80)
        ),
        "ok_label": "End Session",
        "cancel_label": "Keep It",
        "destructive": true,
    });
    let answer = ask_user(state, block, question).await?;
    if answer.answered && answer.approve {
        Ok(())
    } else {
        Err("kept: the user chose not to end it".to_string())
    }
}

fn str_field(data: &serde_json::Value, key: &str) -> String {
    data.get(key)
        .and_then(|v| v.as_str())
        .unwrap_or_default()
        .to_string()
}

/// What `connensure` does: accept `local`; accept a WSL distro that is
/// installed (Windows only), which is then connected; accept an SSH
/// destination where the system `ssh` is, claiming nothing (only a login
/// connects it: a pane's ssh, or `connconnect`); refuse a bad name with the
/// reason, recorded as the status the pane's overlay shows.
pub(crate) async fn ensure(broker: &Broker, name: &str) -> Result<(), String> {
    let (installed, ssh_found) = match ConnTarget::parse(name)? {
        ConnTarget::Wsl(_) => (crate::backend::remote::wsl::list().await, false),
        ConnTarget::Ssh(_) => (Vec::new(), crate::backend::remote::ssh::binary().is_some()),
        ConnTarget::Local => (Vec::new(), false),
    };
    ensure_with(broker, name, &installed, cfg!(windows), ssh_found)
}

/// [`ensure`] with the installed distros, the platform and whether `ssh` was
/// found given, so it is testable anywhere.
pub(crate) fn ensure_with(
    broker: &Broker,
    name: &str,
    wsl_installed: &[String],
    on_windows: bool,
    ssh_found: bool,
) -> Result<(), String> {
    let target = ConnTarget::parse(name)?;
    let message = match &target {
        ConnTarget::Local => return Ok(()),
        ConnTarget::Wsl(_) if !on_windows => "WSL is only available on Windows".to_string(),
        ConnTarget::Wsl(distro) if wsl_installed.iter().any(|d| d.eq_ignore_ascii_case(distro)) => {
            status::set(Some(broker), name, state::CONNECTED, None);
            return Ok(());
        }
        ConnTarget::Wsl(distro) => {
            format!("WSL distribution '{distro}' is not installed (wsl.exe --list shows what is)")
        }
        ConnTarget::Ssh(_) if ssh_found => return Ok(()),
        ConnTarget::Ssh(_) => crate::backend::remote::ssh::missing_binary_message(),
    };
    // Keyed by the name exactly as the pane's meta holds it, not the canonical
    // form: the pane's overlay looks its status up by `meta.connection`, so
    // " area54 " or "host:022" would otherwise never see their error (#4248).
    status::set(Some(broker), name, state::ERROR, Some(&message));
    Err(message)
}

/// Where an agent's `Shell` or `PtyShell` runs.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum AgentTarget {
    Local,
    /// An installed WSL distro.
    Wsl(String),
    /// An SSH destination: runs only with the user's consent
    /// ([`consent_for_ssh`]), and every ssh prompt goes to the user
    /// (`remote::askpass`), never to the agent.
    Ssh(crate::backend::remote::conn::SshDest),
}

/// The connection an agent's `Shell` or `PtyShell` call asked for, checked the
/// same way a pane's is ([`ensure`]), `Err` with the reason when it cannot run.
pub(crate) async fn for_agent(broker: &Broker, name: Option<&str>) -> Result<AgentTarget, String> {
    let name = name.unwrap_or_default();
    ensure(broker, name).await?;
    Ok(match ConnTarget::parse(name)? {
        ConnTarget::Local => AgentTarget::Local,
        ConnTarget::Wsl(distro) => AgentTarget::Wsl(distro),
        ConnTarget::Ssh(dest) => AgentTarget::Ssh(dest),
    })
}

/// How a dialog names an agent: its id, as one plain capped line.
pub(crate) fn agent_label(agent: Option<&str>) -> String {
    agent
        .map(|a| one_line(a, 60))
        .unwrap_or_else(|| "An agent with no AgentMux id".to_string())
}

/// One line of agent-supplied text for a dialog: shown as plain text, newlines
/// folded and length capped, so it cannot lay out a message of its own.
pub(crate) fn one_line(text: &str, max: usize) -> String {
    let flat: String = text
        .chars()
        .map(|c| if c.is_control() { ' ' } else { c })
        .collect();
    let flat = flat.split_whitespace().collect::<Vec<_>>().join(" ");
    if flat.chars().count() <= max {
        flat
    } else {
        format!("{}…", flat.chars().take(max).collect::<String>())
    }
}

/// How long the user has to answer a consent or askpass dialog.
const DIALOG_TIMEOUT_MS: u64 = 120_000;

/// How long an ssh run whose prompts go to the approval window may take: the
/// dialog's own wait, and some to spare.
const SSH_WITH_PROMPTS: std::time::Duration = std::time::Duration::from_millis(DIALOG_TIMEOUT_MS + 30_000);

/// The user's answer to [`ask_user`]. `answered` is false when the window was
/// closed, timed out, or could not be opened.
#[derive(Debug, Default, Clone, serde::Deserialize)]
pub(crate) struct UserAnswer {
    #[serde(default)]
    pub answered: bool,
    #[serde(default)]
    pub approve: bool,
    /// For a `secret` question: never logged.
    #[serde(default)]
    pub text: String,
    #[serde(default)]
    pub checkbox: bool,
}

/// The host that can put a question to the user and relay their answer. Only
/// the host can: agents hold the key srv's own routes take (see `ask_user`).
/// With none connected (headless) there is no one to ask, and whatever needs
/// the answer must not happen, nor wait for one.
pub(crate) async fn host_to_ask_user(state: &AppState) -> Result<crate::server::HostIpc, String> {
    state
        .host_ipc
        .lock()
        .await
        .clone()
        .ok_or_else(|| "no AgentMux window is connected to ask the user in".to_string())
}

/// Ask the user a question about the agent in `agent_block_id`, in an approval
/// subwindow the host opens over the window showing it (crates/cef
/// `ssh_approval`), and wait for the answer.
///
/// Through the host's own IPC server and its token, which no agent holds, and
/// answered in a window the browser API never resolves a pane into: an agent
/// can neither click this answer nor forge it, unlike a modal in the main
/// window answered through srv's own services (both reachable with the auth
/// key every agent has). With no host connected (headless) there is no one to
/// ask: `Err`, so whatever needed the answer does not happen.
pub(crate) async fn ask_user(
    state: &AppState,
    agent_block_id: &str,
    question: serde_json::Value,
) -> Result<UserAnswer, String> {
    let host = host_to_ask_user(state).await?;
    // The agent's pane waits on the user until this returns: the waiting tone
    // and notification for it (REPORT_AGENT_ATTENTION_CTA_CONTRAST_AND_TONE_2026_10_10.md §2).
    let title = question.get("title").and_then(|t| t.as_str()).unwrap_or("An agent needs your answer").to_string();
    let _asking = crate::backend::user_attention::Asking::start(&state.broker, &state.mstore, agent_block_id, "consent", &title);
    let mut body = question;
    body["block_id"] = serde_json::json!(agent_block_id);
    body["timeout_ms"] = serde_json::json!(DIALOG_TIMEOUT_MS);
    let url = format!("http://127.0.0.1:{}/agentmux/approval/ask", host.port);
    let resp = state
        .http_client
        .post(&url)
        .header("Authorization", format!("Bearer {}", host.token))
        .timeout(std::time::Duration::from_millis(DIALOG_TIMEOUT_MS + 15_000))
        .json(&body)
        .send()
        .await
        .map_err(|e| format!("could not ask the user: {e}"))?;
    if !resp.status().is_success() {
        return Err(format!(
            "could not ask the user: the window host answered HTTP {}",
            resp.status()
        ));
    }
    let v: serde_json::Value = resp
        .json()
        .await
        .map_err(|e| format!("could not ask the user: {e}"))?;
    if v.get("ok").and_then(|o| o.as_bool()) != Some(true) {
        let e = v
            .get("error")
            .and_then(|e| e.as_str())
            .unwrap_or("no answer");
        return Err(format!("could not ask the user: {e}"));
    }
    serde_json::from_value(v.get("data").cloned().unwrap_or_default())
        .map_err(|e| format!("could not ask the user: {e}"))
}

/// The agent an SSH request comes from, proven: its signed identity
/// (`UiAutomationAuth`, made with its own jekt key, which no other agent
/// holds) verified, and the pane derived from it, never taken from the
/// request. `Err` unless that pane is `agent_block_id`: every agent holds
/// srv's auth key, so a request body alone could name another agent's pane
/// and borrow its "Always allow".
pub(crate) fn verified_agent(
    state: &AppState,
    agent_block_id: &str,
    auth: Option<&agentmux_common::api_types::UiAutomationAuth>,
) -> Result<String, String> {
    let auth = auth.ok_or_else(|| {
        "an SSH connection needs the agent's signed identity, which this call did not carry (respawn the agent to give it a signing key)".to_string()
    })?;
    let pane = crate::server::ui_handlers::verified_block_id(state, None, auth)
        .map_err(|e| format!("could not verify which agent is asking: {e}"))?;
    if pane != agent_block_id {
        return Err("this request names another agent's pane".to_string());
    }
    Ok(auth.agent_id.clone())
}

/// The user's consent for the agent `agent_id` (verified, [`verified_agent`])
/// in `agent_block_id` to run `what` on the SSH `connection` with the user's
/// identity (spec §8.2). `connection` is the canonical name
/// (`ConnTarget::name`), so one host is one consent however it is spelled.
/// "Always" for that agent and host is remembered (`remote::agent_access`)
/// and asks nothing again; otherwise the user is asked ([`ask_user`]).
pub(crate) async fn consent_for_ssh(
    state: &AppState,
    agent_block_id: &str,
    agent_id: &str,
    connection: &str,
    what: &str,
) -> Result<(), String> {
    use crate::backend::remote::agent_access;
    let agent_id = Some(agent_id.to_string()).filter(|a| !a.trim().is_empty());
    let agent = agent_label(agent_id.as_deref());
    let connection = connection.trim();
    if agent_id
        .as_deref()
        .is_some_and(|id| agent_access::always_allowed(id, connection))
    {
        tracing::info!(agent = %agent, connection = %connection, what = %one_line(what, 200), "agent ssh access: allowed (always)");
        return Ok(());
    }
    let question = serde_json::json!({
        "kind": "consent",
        "title": format!("Agent access to {}", one_line(connection, 80)),
        "message": format!(
            "{agent} wants to run this on {}, as you, with your SSH keys:\n\n{}\n\nAllow it?",
            one_line(connection, 80),
            one_line(what, 400)
        ),
        "checkbox": match &agent_id {
            Some(_) => format!("Always allow {agent} on {}", one_line(connection, 80)),
            None => String::new(),
        },
        "ok_label": "Allow",
        "cancel_label": "Deny",
    });
    let answer = ask_user(state, agent_block_id, question)
        .await
        .map_err(|e| format!("{agent} needs the user's permission to use {connection}, and {e}"))?;
    if !(answer.answered && answer.approve) {
        tracing::info!(agent = %agent, connection = %connection, answered = answer.answered, "agent ssh access: not allowed");
        return Err(if answer.answered {
            format!("the user denied {agent} access to {connection}")
        } else {
            format!("the user did not answer whether {agent} may use {connection}")
        });
    }
    let always = answer.checkbox && agent_id.is_some();
    if let Some(id) = agent_id.as_deref().filter(|_| always) {
        if let Err(e) = agent_access::remember(id, connection) {
            tracing::warn!(error = %e, "agent ssh access: could not remember 'always'");
        }
    }
    tracing::info!(agent = %agent, connection = %connection, what = %one_line(what, 200), always, "agent ssh access: allowed");
    Ok(())
}

/// `POST /api/v1/askpass` body, from `agentmux-bashwrap` as ssh's askpass.
#[derive(Debug, serde::Deserialize)]
pub(crate) struct AskpassRequest {
    secret: String,
    prompt: String,
    /// ssh's `SSH_ASKPASS_PROMPT`: `confirm`, `none`, or empty.
    #[serde(default)]
    hint: String,
}

/// `POST /api/v1/askpass`: show an ssh prompt to the user ([`ask_user`]) and
/// return the answer (`remote::askpass`). Only a live secret is answered, and
/// the secret, not anything in the request, says which agent and host the
/// question names.
pub(crate) async fn handle_askpass(
    axum::extract::State(state): axum::extract::State<AppState>,
    axum::Json(req): axum::Json<AskpassRequest>,
) -> axum::response::Response {
    use crate::backend::remote::askpass::{self, PromptKind};
    use axum::http::StatusCode;
    use axum::response::IntoResponse;
    let refuse = |code: StatusCode, error: String| {
        (code, axum::Json(serde_json::json!({ "error": error }))).into_response()
    };
    let Some(grant) = askpass::lookup(&req.secret) else {
        return refuse(StatusCode::FORBIDDEN, "no such ssh prompt".to_string());
    };
    let kind = askpass::classify(&req.prompt, &req.hint);
    // The grant's names are AgentMux's own, shown as plain capped lines all
    // the same; ssh's prompt is ssh's text.
    let question = serde_json::json!({
        "kind": match kind {
            PromptKind::YesNo => "yesno",
            PromptKind::Info => "info",
            PromptKind::Secret => "secret",
        },
        "title": format!("SSH: {}", one_line(&grant.connection, 80)),
        "message": format!(
            "ssh, connecting to {} for {}, {}:\n\n{}",
            one_line(&grant.connection, 80),
            if grant.user_pane {
                "your terminal".to_string()
            } else {
                agent_label(Some(grant.agent.as_str()).filter(|a| !a.is_empty()))
            },
            if kind == PromptKind::Info { "says" } else { "asks" },
            req.prompt.trim().chars().take(2000).collect::<String>()
        ),
        "ok_label": match kind {
            PromptKind::YesNo => "Yes",
            PromptKind::Info => "OK",
            PromptKind::Secret => "OK",
        },
        "cancel_label": if kind == PromptKind::YesNo { "No" } else { "Cancel" },
    });
    // A notice ("touch your security key") is waited on like a question, but
    // ssh does not wait for it: it kills askpass once done. That drops this
    // request, which drops the request to the host, whose cleanup closes the
    // window, so the notice goes exactly when ssh is done with it.
    match ask_user(&state, &grant.agent_block_id, question).await {
        Ok(answer) if answer.answered => {
            // Never logged: the answer may be a password.
            let text = match kind {
                PromptKind::YesNo if answer.approve => "yes".to_string(),
                PromptKind::YesNo => "no".to_string(),
                PromptKind::Secret if answer.approve => answer.text,
                PromptKind::Secret => {
                    return refuse(StatusCode::GONE, "cancelled".to_string());
                }
                PromptKind::Info => String::new(),
            };
            axum::Json(serde_json::json!({ "answer": text })).into_response()
        }
        Ok(_) => refuse(StatusCode::GONE, "not answered".to_string()),
        Err(e) => refuse(StatusCode::CONFLICT, e),
    }
}

/// One row of the agent `ConnList` tool.
#[derive(Debug, Clone, PartialEq, serde::Serialize)]
pub(crate) struct ConnEntry {
    pub connection: String,
    /// `local`, `wsl` or `ssh`.
    pub kind: &'static str,
    /// `connected`, `available` (usable, not opened yet), or a status from
    /// `remote::status` (`connecting`, `disconnected`, `error`).
    pub status: String,
    #[serde(skip_serializing_if = "String::is_empty")]
    pub error: String,
    /// Whether an agent can pass it as `connection`: every connection AgentMux
    /// can run. An SSH host also asks the user for consent the first time
    /// ([`consent_for_ssh`]).
    pub agent_can_use: bool,
}

/// The picker's remote names: the ssh config hosts, then any other SSH
/// connection known this session, each once.
pub(crate) fn remote_names(ssh_hosts: &[String], known: &[ConnStatus]) -> Vec<String> {
    use crate::backend::remote::conn::same_connection;
    let mut out: Vec<String> = ssh_hosts.to_vec();
    for s in known {
        let is_ssh = matches!(ConnTarget::parse(&s.connection), Ok(ConnTarget::Ssh(_)));
        if is_ssh && !out.iter().any(|n| same_connection(n, &s.connection)) {
            out.push(s.connection.clone());
        }
    }
    out
}

/// The connections there are: this machine, each installed WSL distro, the
/// user's ssh config hosts, and any other connection a pane has used this
/// session, with its status and whether an agent can use it.
pub(crate) fn list(
    wsl_installed: &[String],
    ssh_hosts: &[String],
    known: &[ConnStatus],
) -> Vec<ConnEntry> {
    use crate::backend::remote::conn::same_connection;
    let status_of = |name: &str| known.iter().find(|s| same_connection(&s.connection, name));
    let mut out = vec![ConnEntry {
        connection: "local".to_string(),
        kind: "local",
        status: state::CONNECTED.to_string(),
        error: String::new(),
        agent_can_use: true,
    }];
    for distro in wsl_installed {
        let name = ConnTarget::Wsl(distro.clone()).name();
        let (status, error) = match status_of(&name) {
            Some(s) => (s.status.clone(), s.error.clone()),
            None => ("available".to_string(), String::new()),
        };
        out.push(ConnEntry {
            connection: name,
            kind: "wsl",
            status,
            error,
            agent_can_use: true,
        });
    }
    for host in ssh_hosts {
        let (status, error) = match status_of(host) {
            Some(s) => (s.status.clone(), s.error.clone()),
            None => ("available".to_string(), String::new()),
        };
        out.push(ConnEntry {
            connection: host.clone(),
            kind: "ssh",
            status,
            error,
            agent_can_use: true,
        });
    }
    for s in known {
        if out
            .iter()
            .any(|e| same_connection(&e.connection, &s.connection))
        {
            continue;
        }
        let kind = match ConnTarget::parse(&s.connection) {
            Ok(ConnTarget::Wsl(_)) => "wsl",
            Ok(ConnTarget::Ssh(_)) => "ssh",
            _ => continue,
        };
        out.push(ConnEntry {
            connection: s.connection.clone(),
            kind,
            status: s.status.clone(),
            error: s.error.clone(),
            agent_can_use: true,
        });
    }
    out
}

/// `GET /api/v1/conn/list`: the agent `ConnList` tool (spec §8.1).
pub(crate) async fn handle_conn_list() -> axum::Json<Vec<ConnEntry>> {
    let installed = crate::backend::remote::wsl::list().await;
    let hosts = tokio::task::spawn_blocking(crate::backend::remote::ssh_config::hosts)
        .await
        .unwrap_or_default();
    axum::Json(list(&installed, &hosts, &status::all()))
}

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

    #[test]
    fn local_is_accepted_and_remote_is_refused_with_a_status() {
        let broker = Broker::new();
        let none: &[String] = &[];
        let ensure = |name: &str| ensure_with(&broker, name, none, true, false);
        assert_eq!(ensure("local"), Ok(()));
        assert_eq!(ensure(""), Ok(()));

        let err = ensure("test-ensure-host").unwrap_err();
        assert!(err.contains("need the ssh command"), "{err}");
        // With ssh: accepted, and nothing claimed. Having an ssh command is
        // not a login: the host showed connected before anything had
        // connected (#4513). A pane's ssh, or a login, says how it is.
        assert_eq!(
            ensure_with(&broker, "test-ensure-ssh-ok", none, true, true),
            Ok(())
        );
        assert!(!status::all()
            .iter()
            .any(|s| s.connection == "test-ensure-ssh-ok" && s.connected));
        let st = status::all()
            .into_iter()
            .find(|s| s.connection == "test-ensure-host")
            .unwrap();
        assert_eq!(
            (st.status.as_str(), st.error.as_str()),
            (state::ERROR, err.as_str())
        );

        // A name that is not a connection says why, and records nothing.
        assert!(ensure("-oProxyCommand=calc")
            .unwrap_err()
            .contains("cannot start with '-'"));
        assert!(!status::all().iter().any(|s| s.connection.starts_with('-')));
    }

    #[test]
    fn an_installed_wsl_distro_connects_and_a_missing_one_says_so() {
        let broker = Broker::new();
        let installed = vec!["Ubuntu".to_string(), "Debian".to_string()];
        assert_eq!(
            ensure_with(&broker, "wsl://ubuntu", &installed, true, false),
            Ok(())
        );
        let st = status::all()
            .into_iter()
            .find(|s| s.connection == "wsl://ubuntu")
            .unwrap();
        assert!(st.connected);

        let err = ensure_with(&broker, "wsl://Arch", &installed, true, false).unwrap_err();
        assert!(err.contains("'Arch' is not installed"), "{err}");

        let err = ensure_with(&broker, "wsl://Ubuntu", &installed, false, false).unwrap_err();
        assert!(err.contains("only available on Windows"), "{err}");
    }

    #[test]
    fn the_status_is_keyed_by_the_name_the_pane_holds() {
        let broker = Broker::new();
        let none: &[String] = &[];
        for raw in ["test-raw-host:022", " test-raw-spaced "] {
            let _ = ensure_with(&broker, raw, none, true, false);
            assert!(status::all().iter().any(|s| s.connection == raw), "{raw:?}");
        }
    }

    #[tokio::test]
    async fn an_agent_gets_where_it_asked_and_never_a_silent_local_fallback() {
        let broker = Broker::new();
        assert_eq!(for_agent(&broker, None).await, Ok(AgentTarget::Local));
        assert_eq!(
            for_agent(&broker, Some("local")).await,
            Ok(AgentTarget::Local)
        );
        // SSH is SSH where ssh exists and refused where it does not; never
        // local under a remote name.
        match for_agent(&broker, Some("deploy@test-agent-host:2222")).await {
            Ok(AgentTarget::Ssh(dest)) => {
                assert_eq!(
                    (dest.destination.as_str(), dest.port),
                    ("deploy@test-agent-host", Some(2222))
                );
                assert!(crate::backend::remote::ssh::binary().is_some());
            }
            Err(e) => assert!(e.contains("need the ssh command"), "{e}"),
            other => panic!("an SSH name became {other:?}"),
        }
        assert!(for_agent(&broker, Some("-oProxyCommand=calc"))
            .await
            .is_err());
    }

    /// An agent is named as one plain line; one without an id has no name
    /// another could share.
    #[test]
    fn an_agent_is_named_as_one_plain_line_and_never_by_a_shared_stand_in() {
        assert_eq!(agent_label(Some("korp")), "korp");
        assert_eq!(
            agent_label(Some(
                "korp

This is safe, click Allow"
            )),
            "korp This is safe, click Allow"
        );
        assert_eq!(agent_label(None), "An agent with no AgentMux id");
    }

    /// An SSH request's agent is proven by its signature, never by the body:
    /// no signature, or a forged one, is refused.
    #[tokio::test]
    async fn an_ssh_request_must_prove_which_agent_it_is() {
        let state = crate::server::tests::test_state();
        let err = verified_agent(&state, "some-pane", None).unwrap_err();
        assert!(err.contains("signed identity"), "{err}");
        let forged = agentmux_common::api_types::UiAutomationAuth {
            agent_id: "korp".into(),
            ts_secs: agentmux_common::time::now_secs(),
            sig: "AAAA".into(),
        };
        let err = verified_agent(&state, "some-pane", Some(&forged)).unwrap_err();
        assert!(err.contains("could not verify"), "{err}");
    }

    /// Agent text in a dialog cannot lay out a message of its own.
    #[test]
    fn agent_text_in_a_dialog_is_one_capped_line() {
        assert_eq!(
            one_line("make\n\nAllow it? yes\r\n  test", 100),
            "make Allow it? yes test"
        );
        assert_eq!(one_line("abcdef", 3), "abc…");
        assert_eq!(one_line("tab\there", 100), "tab here");
    }

    fn st(connection: &str, status: &str, error: &str) -> ConnStatus {
        ConnStatus {
            status: status.to_string(),
            connection: connection.to_string(),
            error: error.to_string(),
            ..Default::default()
        }
    }

    #[test]
    fn the_list_is_local_then_the_distros_then_the_ssh_hosts_then_what_panes_have_used() {
        let installed = vec!["Ubuntu".to_string(), "Debian".to_string()];
        let hosts = vec!["area54".to_string(), "nas".to_string()];
        let known = vec![
            // Accepted as typed; still the installed Ubuntu, listed once.
            st("wsl://ubuntu", state::CONNECTED, ""),
            st("area54", state::ERROR, "could not connect"),
            st("deploy@10.0.0.5", state::CONNECTED, ""),
            st("wsl://Arch", state::ERROR, "not installed"),
            // Not a connection: never listed.
            st("local", state::CONNECTED, ""),
        ];
        let got: Vec<(String, &str, String, bool)> = list(&installed, &hosts, &known)
            .into_iter()
            .map(|e| (e.connection, e.kind, e.status, e.agent_can_use))
            .collect();
        let want = [
            ("local", "local", "connected", true),
            ("wsl://Ubuntu", "wsl", "connected", true),
            ("wsl://Debian", "wsl", "available", true),
            ("area54", "ssh", "error", true),
            ("nas", "ssh", "available", true),
            ("deploy@10.0.0.5", "ssh", "connected", true),
            ("wsl://Arch", "wsl", "error", true),
        ];
        let want: Vec<(String, &str, String, bool)> = want
            .iter()
            .map(|(c, k, s, a)| (c.to_string(), *k, s.to_string(), *a))
            .collect();
        assert_eq!(got, want);
    }

    /// The picker's "Remote" section: config hosts first, then other SSH
    /// connections used this session, never WSL or local.
    #[test]
    fn the_picker_gets_the_ssh_hosts_and_the_ssh_connections_used() {
        let hosts = vec!["area54".to_string(), "nas".to_string()];
        let known = vec![
            st("area54", state::CONNECTED, ""),
            st("deploy@10.0.0.5:2222", state::DISCONNECTED, ""),
            st("wsl://Ubuntu", state::CONNECTED, ""),
            st("local", state::CONNECTED, ""),
        ];
        assert_eq!(
            remote_names(&hosts, &known),
            ["area54", "nas", "deploy@10.0.0.5:2222"]
        );
    }
}