feat: manage multiple ssh machines from one client (#3670)

5 files changed, 1060 insertions, 399 deletions

diff --git a/src/remote/args.rs b/src/remote/args.rs
new file mode 100644
index 00000000..476acf70
--- /dev/null
+++ b/src/remote/args.rs
@@ -0,0 +1,126 @@
+pub(crate) const REATTACH_COMMAND_ENV_VAR: &str = "HERDR_REATTACH_COMMAND";
+pub(crate) const REMOTE_KEYBINDINGS_ENV_VAR: &str = "HERDR_REMOTE_KEYBINDINGS";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum RemoteKeybindings {
+    Local,
+    Server,
+}
+
+impl RemoteKeybindings {
+    pub(super) fn parse(value: &str) -> Result<Self, String> {
+        match value {
+            "local" => Ok(Self::Local),
+            "server" => Ok(Self::Server),
+            _ => Err("--remote-keybindings must be 'local' or 'server'".to_string()),
+        }
+    }
+
+    pub(super) fn as_str(self) -> &'static str {
+        match self {
+            Self::Local => "local",
+            Self::Server => "server",
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct RemoteLaunch {
+    pub(crate) target: String,
+    pub(crate) keybindings: RemoteKeybindings,
+    pub(crate) live_handoff: bool,
+}
+
+pub(crate) fn extract_remote_args(
+    args: &[String],
+) -> Result<(Vec<String>, Option<RemoteLaunch>), String> {
+    let mut cleaned = Vec::with_capacity(args.len());
+    if let Some(program) = args.first() {
+        cleaned.push(program.clone());
+    }
+
+    let mut remote_target = None;
+    let mut keybindings = RemoteKeybindings::Local;
+    let mut keybindings_seen = false;
+    let mut live_handoff = false;
+    let mut index = 1;
+    while index < args.len() {
+        let arg = &args[index];
+        if arg == "--" {
+            cleaned.extend_from_slice(&args[index..]);
+            break;
+        }
+        if arg == "--handoff" {
+            live_handoff = true;
+            index += 1;
+            continue;
+        }
+        if arg == "--remote" {
+            if remote_target.is_some() {
+                return Err("--remote can only be specified once".to_string());
+            }
+            let Some(value) = args.get(index + 1) else {
+                return Err("missing value for --remote".to_string());
+            };
+            remote_target = Some(validate_remote_target(value)?.to_owned());
+            index += 2;
+            continue;
+        }
+        if let Some(value) = arg.strip_prefix("--remote=") {
+            if remote_target.is_some() {
+                return Err("--remote can only be specified once".to_string());
+            }
+            remote_target = Some(validate_remote_target(value)?.to_owned());
+            index += 1;
+            continue;
+        }
+        if arg == "--remote-keybindings" {
+            if keybindings_seen {
+                return Err("--remote-keybindings can only be specified once".to_string());
+            }
+            let Some(value) = args.get(index + 1) else {
+                return Err("missing value for --remote-keybindings".to_string());
+            };
+            keybindings = RemoteKeybindings::parse(value)?;
+            keybindings_seen = true;
+            index += 2;
+            continue;
+        }
+        if let Some(value) = arg.strip_prefix("--remote-keybindings=") {
+            if keybindings_seen {
+                return Err("--remote-keybindings can only be specified once".to_string());
+            }
+            keybindings = RemoteKeybindings::parse(value)?;
+            keybindings_seen = true;
+            index += 1;
+            continue;
+        }
+
+        cleaned.push(arg.clone());
+        index += 1;
+    }
+
+    let remote = remote_target.map(|target| RemoteLaunch {
+        target,
+        keybindings,
+        live_handoff,
+    });
+    if remote.is_none() && keybindings_seen {
+        return Err("--remote-keybindings requires --remote".to_string());
+    }
+    if remote.is_none() && live_handoff {
+        cleaned.push("--handoff".to_string());
+    }
+
+    Ok((cleaned, remote))
+}
+
+pub(crate) fn validate_remote_target(target: &str) -> Result<&str, String> {
+    if target.is_empty() {
+        return Err("missing value for --remote".to_string());
+    }
+    if target.starts_with('-') {
+        return Err("--remote target must not start with '-'".to_string());
+    }
+    Ok(target)
+}
diff --git a/src/remote/attach.rs b/src/remote/attach.rs
index 641aad00..81810411 100644
--- a/src/remote/attach.rs
+++ b/src/remote/attach.rs
@@ -1,15 +1,13 @@
 //! Remote thin-client launcher over SSH command stdio.
 
-use super::shell_quote;
+use super::{args::*, process::wait_with_output_timeout, restart_policy::*, shell_quote};
 use std::collections::BTreeMap;
 use std::fs::{self, File};
 use std::io::{self, IsTerminal, Write as _};
 use std::path::{Path, PathBuf};
 use std::process::{Command, Output, Stdio};
 
-use interprocess::local_socket::traits::Listener as _;
-#[cfg(windows)]
-use interprocess::local_socket::traits::Stream as _;
+use interprocess::local_socket::traits::{Listener as _, Stream as _};
 use interprocess::local_socket::ListenerNonblockingMode;
 use interprocess::TryClone as _;
 use serde::Deserialize;
@@ -21,144 +19,16 @@ use std::thread::{self, JoinHandle};
 use std::time::{Duration, Instant};
 
 const BRIDGE_ACCEPT_POLL: Duration = Duration::from_millis(50);
-#[cfg(windows)]
 const BRIDGE_IO_POLL: Duration = Duration::from_millis(1);
 const BRIDGE_SOCKET_PERMISSION_MODE: u32 = 0o600;
 const REMOTE_SERVER_SHUTDOWN_CONFIRM_TIMEOUT: Duration = Duration::from_secs(5);
+const NONINTERACTIVE_SSH_COMMAND_TIMEOUT: Duration = Duration::from_secs(15);
 const REMOTE_SERVER_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(100);
 const CURRENT_PROTOCOL: u32 = crate::protocol::PROTOCOL_VERSION;
 const STABLE_UPDATE_MANIFEST_URL: &str = "https://herdr.dev/latest.json";
 const PREVIEW_UPDATE_MANIFEST_URL: &str = "https://herdr.dev/preview.json";
 const REMOTE_BINARY_ENV_VAR: &str = "HERDR_REMOTE_BINARY";
 const SSH_CONTROL_SOCKET_NAME: &str = "ctl";
-pub(crate) const REATTACH_COMMAND_ENV_VAR: &str = "HERDR_REATTACH_COMMAND";
-
-pub(crate) const REMOTE_KEYBINDINGS_ENV_VAR: &str = "HERDR_REMOTE_KEYBINDINGS";
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(crate) enum RemoteKeybindings {
-    Local,
-    Server,
-}
-
-impl RemoteKeybindings {
-    fn parse(value: &str) -> Result<Self, String> {
-        match value {
-            "local" => Ok(Self::Local),
-            "server" => Ok(Self::Server),
-            _ => Err("--remote-keybindings must be 'local' or 'server'".to_string()),
-        }
-    }
-
-    fn as_str(self) -> &'static str {
-        match self {
-            Self::Local => "local",
-            Self::Server => "server",
-        }
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub(crate) struct RemoteLaunch {
-    pub(crate) target: String,
-    pub(crate) keybindings: RemoteKeybindings,
-    pub(crate) live_handoff: bool,
-}
-
-pub(crate) fn extract_remote_args(
-    args: &[String],
-) -> Result<(Vec<String>, Option<RemoteLaunch>), String> {
-    let mut cleaned = Vec::with_capacity(args.len());
-    if let Some(program) = args.first() {
-        cleaned.push(program.clone());
-    }
-
-    let mut remote_target = None;
-    let mut keybindings = RemoteKeybindings::Local;
-    let mut keybindings_seen = false;
-    let mut live_handoff = false;
-    let mut index = 1;
-    while index < args.len() {
-        let arg = &args[index];
-        if arg == "--" {
-            cleaned.extend_from_slice(&args[index..]);
-            break;
-        }
-        if arg == "--handoff" {
-            live_handoff = true;
-            index += 1;
-            continue;
-        }
-        if arg == "--remote" {
-            if remote_target.is_some() {
-                return Err("--remote can only be specified once".to_string());
-            }
-            let Some(value) = args.get(index + 1) else {
-                return Err("missing value for --remote".to_string());
-            };
-            remote_target = Some(validate_remote_target(value)?.to_owned());
-            index += 2;
-            continue;
-        }
-        if let Some(value) = arg.strip_prefix("--remote=") {
-            if remote_target.is_some() {
-                return Err("--remote can only be specified once".to_string());
-            }
-            remote_target = Some(validate_remote_target(value)?.to_owned());
-            index += 1;
-            continue;
-        }
-        if arg == "--remote-keybindings" {
-            if keybindings_seen {
-                return Err("--remote-keybindings can only be specified once".to_string());
-            }
-            let Some(value) = args.get(index + 1) else {
-                return Err("missing value for --remote-keybindings".to_string());
-            };
-            keybindings = RemoteKeybindings::parse(value)?;
-            keybindings_seen = true;
-            index += 2;
-            continue;
-        }
-        if let Some(value) = arg.strip_prefix("--remote-keybindings=") {
-            if keybindings_seen {
-                return Err("--remote-keybindings can only be specified once".to_string());
-            }
-            keybindings = RemoteKeybindings::parse(value)?;
-            keybindings_seen = true;
-            index += 1;
-            continue;
-        }
-
-        cleaned.push(arg.clone());
-        index += 1;
-    }
-
-    let remote = remote_target.map(|target| RemoteLaunch {
-        target,
-        keybindings,
-        live_handoff,
-    });
-    if remote.is_none() && keybindings_seen {
-        return Err("--remote-keybindings requires --remote".to_string());
-    }
-    if remote.is_none() && live_handoff {
-        cleaned.push("--handoff".to_string());
-    }
-
-    Ok((cleaned, remote))
-}
-
-fn validate_remote_target(target: &str) -> Result<&str, String> {
-    if target.is_empty() {
-        return Err("missing value for --remote".to_string());
-    }
-    if target.starts_with('-') {
-        return Err("--remote target must not start with '-'".to_string());
-    }
-    Ok(target)
-}
-
 pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
     let session_name = crate::session::active_name()
         .unwrap_or_else(|| crate::session::DEFAULT_SESSION_NAME.to_string());
@@ -177,13 +47,22 @@ pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
         .config
         .remote
         .manage_ssh_config;
-    let remote_ssh = RemoteSsh::new(remote.target.clone(), manage_ssh_config);
-    let prepared_remote = prepare_remote_herdr(&remote_ssh, remote.live_handoff)?;
+    let require_surface_interest = crate::client::endpoint::EndpointCatalog::load()
+        .map(|catalog| catalog.contains_enabled_target_session(&remote.target, &session_name))
+        .unwrap_or(false);
+    let remote_ssh = RemoteSsh::new(
+        remote.target.clone(),
+        manage_ssh_config,
+        session_name.clone(),
+    );
+    let prepared_remote =
+        prepare_remote_herdr(&remote_ssh, remote.live_handoff, require_surface_interest)?;
     ensure_remote_server_ready(
         &remote_ssh,
         &prepared_remote.remote_herdr,
         prepared_remote.stop_after_install_approved,
         remote.live_handoff,
+        require_surface_interest,
     )?;
 
     let _bridge = SshStdioBridge::start(
@@ -192,11 +71,68 @@ pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
         local_socket.clone(),
         session_name,
         remote_ssh.options(),
+        false,
     )?;
 
     run_client_process(&local_socket, &reattach_command, remote.keybindings)
 }
 
+pub(crate) fn prepare_saved_ssh(target: &str, session_name: &str) -> io::Result<()> {
+    super::validate_remote_target(target)
+        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
+    crate::session::validate_name(session_name)
+        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
+    let manage_ssh_config = crate::config::Config::load()
+        .config
+        .remote
+        .manage_ssh_config;
+    let ssh = RemoteSsh::new(
+        target.to_owned(),
+        manage_ssh_config,
+        session_name.to_owned(),
+    );
+    let prepared = prepare_remote_herdr(&ssh, true, true)?;
+    ensure_remote_server_ready(
+        &ssh,
+        &prepared.remote_herdr,
+        prepared.stop_after_install_approved,
+        true,
+        true,
+    )?;
+
+    // The bridge already owns daemon startup. EOF closes only this temporary attachment,
+    // leaving the named server running even when no local TUI is open yet.
+    let output = ssh.sh_output(&format!(
+        "{} </dev/null",
+        remote_bridge_command(&prepared.remote_herdr, session_name)
+    ))?;
+    if !output.status.success() {
+        return Err(command_failed("remote server startup failed", &output));
+    }
+    match remote_server_status(&ssh, &prepared.remote_herdr, true)? {
+        RemoteServerStatus::Running {
+            endpoint_protocol_generation,
+            surface_interest,
+            health_check,
+            detached_server_daemon,
+            ..
+        } if remote_server_restart_reason(
+            endpoint_protocol_generation,
+            detached_server_daemon,
+            true,
+            surface_interest,
+            health_check,
+        )
+        .is_none() =>
+        {
+            Ok(())
+        }
+        _ => Err(io::Error::other(
+            "remote server is not ready for saved machines",
+        )),
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct RemotePlatform {
     os: &'static str,
@@ -244,7 +180,7 @@ impl RemotePlatform {
 }
 
 #[derive(Debug, Clone)]
-struct RemoteHerdr {
+pub(super) struct RemoteHerdr {
     install_suffix: String,
     shell_path: String,
     platform: RemotePlatform,
@@ -392,13 +328,13 @@ struct RemoteReleaseAsset {
     sha256: Option<String>,
 }
 
-struct PreparedRemoteHerdr {
-    remote_herdr: RemoteHerdr,
+pub(super) struct PreparedRemoteHerdr {
+    pub(super) remote_herdr: RemoteHerdr,
     stop_after_install_approved: bool,
 }
 
 #[derive(Clone)]
-struct ManagedSshOptions {
+pub(super) struct ManagedSshOptions {
     config_path: PathBuf,
     control_path: Option<PathBuf>,
 }
@@ -415,13 +351,15 @@ impl Drop for ManagedSshConfig {
     }
 }
 
-struct RemoteSsh {
+pub(super) struct RemoteSsh {
     target: String,
+    session_name: String,
     managed_config: Option<ManagedSshConfig>,
+    noninteractive: bool,
 }
 
 impl RemoteSsh {
-    fn new(target: String, manage_ssh_config: bool) -> Self {
+    fn new(target: String, manage_ssh_config: bool, session_name: String) -> Self {
         let managed_config = if manage_ssh_config {
             write_managed_ssh_config()
                 .inspect_err(|err| {
@@ -434,7 +372,18 @@ impl RemoteSsh {
 
         Self {
             target,
+            session_name,
             managed_config,
+            noninteractive: false,
+        }
+    }
+
+    pub(super) fn new_noninteractive(target: String) -> Self {
+        Self {
+            target,
+            session_name: crate::session::DEFAULT_SESSION_NAME.into(),
+            managed_config: None,
+            noninteractive: true,
         }
     }
 
@@ -442,12 +391,19 @@ impl RemoteSsh {
         &self.target
     }
 
-    fn options(&self) -> Option<&ManagedSshOptions> {
+    fn destination(&self) -> String {
+        format!("{} (session {})", self.target, self.session_name)
+    }
+
+    pub(super) fn options(&self) -> Option<&ManagedSshOptions> {
         self.managed_config.as_ref().map(|config| &config.options)
     }
 
     fn command(&self) -> Command {
         let mut command = self.base_command();
+        if self.noninteractive {
+            apply_noninteractive_ssh_options(&mut command);
+        }
         command.arg("-T").arg(&self.target);
         command
     }
@@ -475,13 +431,26 @@ impl RemoteSsh {
                 "ssh bootstrap stdin missing",
             ))
         };
-        let output = child.wait_with_output()?;
+        let output = if self.noninteractive {
+            wait_with_output_timeout(child, NONINTERACTIVE_SSH_COMMAND_TIMEOUT)?
+        } else {
+            child.wait_with_output()?
+        };
         write_result?;
         Ok(output)
     }
 
-    fn user_shell_output(&self, command: &str) -> io::Result<Output> {
-        self.command().arg(command).output()
+    fn user_shell_output(&self, remote_command: &str) -> io::Result<Output> {
+        let mut command = self.command();
+        command
+            .arg(remote_command)
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped());
+        if self.noninteractive {
+            wait_with_output_timeout(command.spawn()?, NONINTERACTIVE_SSH_COMMAND_TIMEOUT)
+        } else {
+            command.output()
+        }
     }
 
     fn install_herdr(&self, remote_herdr: &RemoteHerdr, source_path: &Path) -> io::Result<()> {
@@ -601,6 +570,24 @@ impl Drop for RemoteSsh {
     }
 }
 
+fn apply_noninteractive_ssh_options(command: &mut Command) {
+    command
+        .arg("-o")
+        .arg("BatchMode=yes")
+        .arg("-o")
+        .arg("NumberOfPasswordPrompts=0")
+        .arg("-o")
+        .arg("StrictHostKeyChecking=yes")
+        .arg("-o")
+        .arg("ConnectTimeout=10")
+        .arg("-o")
+        .arg("ConnectionAttempts=1")
+        .arg("-o")
+        .arg("ServerAliveInterval=15")
+        .arg("-o")
+        .arg("ServerAliveCountMax=4");
+}
+
 fn apply_managed_ssh_options(command: &mut Command, options: Option<&ManagedSshOptions>) {
     let Some(options) = options else {
         return;
@@ -640,9 +627,10 @@ impl InstallSource {
     }
 }
 
-fn prepare_remote_herdr(
+pub(super) fn prepare_remote_herdr(
     ssh: &RemoteSsh,
     live_handoff_enabled: bool,
+    require_surface_interest: bool,
 ) -> io::Result<PreparedRemoteHerdr> {
     let platform = detect_remote_platform(ssh)?;
     let remote_herdr = RemoteHerdr::for_platform(platform);
@@ -651,14 +639,20 @@ fn prepare_remote_herdr(
 
     if override_binary.is_none() {
         for candidate in &remote_binary_candidates {
-            if remote_binary_supports_endpoint(ssh, candidate).unwrap_or(false) {
+            if remote_binary_supports_endpoint_requirement(ssh, candidate, require_surface_interest)
+                .unwrap_or(false)
+            {
                 return Ok(PreparedRemoteHerdr {
                     remote_herdr: candidate.clone(),
                     stop_after_install_approved: false,
                 });
             }
         }
-        if remote_binary_supports_endpoint(ssh, &remote_herdr)? {
+        if remote_binary_supports_endpoint_requirement(
+            ssh,
+            &remote_herdr,
+            require_surface_interest,
+        )? {
             return Ok(PreparedRemoteHerdr {
                 remote_herdr,
                 stop_after_install_approved: false,
@@ -676,10 +670,11 @@ fn prepare_remote_herdr(
             ssh,
             status_probe_herdr,
             live_handoff_enabled,
+            require_surface_interest,
         )?;
     }
     confirm_remote_install(
-        ssh.target(),
+        &ssh.destination(),
         &remote_herdr,
         &install_source_description(&remote_herdr.platform, override_binary.as_deref()),
     )?;
@@ -688,11 +683,10 @@ fn prepare_remote_herdr(
     source.cleanup();
     install_result?;
 
-    if !remote_binary_supports_endpoint(ssh, &remote_herdr)? {
+    if !remote_binary_supports_endpoint_requirement(ssh, &remote_herdr, require_surface_interest)? {
         return Err(io::Error::other(format!(
-            "installed remote herdr at {}, but it does not support endpoint generation {}",
-            remote_herdr.shell_path,
-            crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
+            "installed remote herdr at {}, but it does not support saved SSH endpoint federation",
+            remote_herdr.shell_path
         )));
     }
     warn_if_remote_bin_not_on_path(ssh)?;
@@ -703,6 +697,25 @@ fn prepare_remote_herdr(
     })
 }
 
+pub(super) fn find_installed_remote_herdr(ssh: &RemoteSsh) -> io::Result<RemoteHerdr> {
+    let platform = detect_remote_platform(ssh)?;
+    let remote_herdr = RemoteHerdr::for_platform(platform);
+    let candidates = remote_binary_candidates(ssh, &remote_herdr)?;
+    for candidate in candidates {
+        if remote_binary_supports_endpoint_requirement(ssh, &candidate, true)? {
+            return Ok(candidate);
+        }
+    }
+    Err(io::Error::new(
+        io::ErrorKind::Unsupported,
+        format!(
+            "matching Herdr is not ready on {}; run `herdr --remote {}` interactively to install or update it",
+            ssh.target(),
+            ssh.target()
+        ),
+    ))
+}
+
 fn detect_remote_platform(ssh: &RemoteSsh) -> io::Result<RemotePlatform> {
     let output = ssh.sh_output("uname -s\nuname -m\n")?;
     if !output.status.success() {
@@ -869,6 +882,9 @@ fn remote_client_status(
     );
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
+        if output.status.code() == Some(255) {
+            return Err(command_failed("remote SSH connection failed", &output));
+        }
         return Ok(None);
     }
     Ok(parse_client_status_json(&String::from_utf8_lossy(
@@ -876,13 +892,13 @@ fn remote_client_status(
     )))
 }
 
-fn remote_binary_supports_endpoint(
+fn remote_binary_supports_endpoint_requirement(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
+    require_surface_interest: bool,
 ) -> io::Result<bool> {
     Ok(remote_client_status(ssh, remote_herdr)?
-        .and_then(|status| status.endpoint_protocol_generation)
-        == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION))
+        .is_some_and(|status| status.supports_endpoint_requirement(require_surface_interest)))
 }
 
 fn remote_binary_exists(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<bool> {
@@ -986,23 +1002,30 @@ enum RemoteServerStatus {
     Running {
         version: Option<String>,
         endpoint_protocol_generation: Option<u32>,
+        surface_interest: bool,
+        health_check: bool,
         live_handoff: bool,
         detached_server_daemon: bool,
     },
     NotRunning,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum RemoteServerRestartReason {
-    EndpointProtocolMissing,
-    DaemonDetachMissing,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum RemoteInstallRunningServerPlan {
-    KeepRunning,
-    LiveHandoff,
-    StopRequired(RemoteServerRestartReason),
+impl RemoteServerStatus {
+    fn with_endpoint_negotiation(
+        mut self,
+        negotiation: &crate::client::endpoint::EndpointNegotiation,
+    ) -> Self {
+        if let Self::Running {
+            surface_interest,
+            health_check,
+            ..
+        } = &mut self
+        {
+            *surface_interest = negotiation.supports_surface_interest();
+            *health_check = negotiation.supports_health_check();
+        }
+        self
+    }
 }
 
 fn ensure_remote_server_ready(
@@ -1010,11 +1033,14 @@ fn ensure_remote_server_ready(
     remote_herdr: &RemoteHerdr,
     stop_after_install_approved: bool,
     live_handoff_enabled: bool,
+    require_surface_interest: bool,
 ) -> io::Result<()> {
-    let status = remote_server_status(ssh, remote_herdr)?;
+    let status = remote_server_status(ssh, remote_herdr, require_surface_interest)?;
     let RemoteServerStatus::Running {
         version,
         endpoint_protocol_generation,
+        surface_interest,
+        health_check,
         live_handoff,
         detached_server_daemon,
     } = status
@@ -1022,9 +1048,13 @@ fn ensure_remote_server_ready(
         return Ok(());
     };
 
-    let Some(reason) =
-        remote_server_restart_reason(endpoint_protocol_generation, detached_server_daemon)
-    else {
+    let Some(reason) = remote_server_restart_reason(
+        endpoint_protocol_generation,
+        detached_server_daemon,
+        require_surface_interest,
+        surface_interest,
+        health_check,
+    ) else {
         return Ok(());
     };
 
@@ -1043,33 +1073,20 @@ fn ensure_remote_server_ready(
         return Ok(());
     }
 
-    if confirm_remote_server_stop(ssh.target(), version.as_deref(), reason)? {
+    if confirm_remote_server_stop(&ssh.destination(), version.as_deref(), reason)? {
         stop_remote_server(ssh, remote_herdr)?;
     }
     Ok(())
 }
 
-fn remote_server_restart_reason(
-    endpoint_protocol_generation: Option<u32>,
-    detached_server_daemon: bool,
-) -> Option<RemoteServerRestartReason> {
-    if endpoint_protocol_generation != Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION)
-    {
-        return Some(RemoteServerRestartReason::EndpointProtocolMissing);
-    }
-    if !detached_server_daemon {
-        return Some(RemoteServerRestartReason::DaemonDetachMissing);
-    }
-    None
-}
-
 fn confirm_remote_install_with_running_server(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
     live_handoff_enabled: bool,
+    require_surface_interest: bool,
 ) -> io::Result<bool> {
-    let target = ssh.target();
-    let status = match remote_server_status(ssh, remote_herdr) {
+    let target = ssh.destination();
+    let status = match remote_server_status(ssh, remote_herdr, require_surface_interest) {
         Ok(status) => status,
         Err(err) => {
             if !io::stdin().is_terminal() {
@@ -1098,6 +1115,8 @@ fn confirm_remote_install_with_running_server(
     let RemoteServerStatus::Running {
         version,
         endpoint_protocol_generation,
+        surface_interest,
+        health_check,
         live_handoff,
         detached_server_daemon,
     } = &status
@@ -1107,8 +1126,11 @@ fn confirm_remote_install_with_running_server(
     let plan = remote_install_running_server_plan(
         *endpoint_protocol_generation,
         *detached_server_daemon,
+        *surface_interest,
+        *health_check,
         *live_handoff,
         live_handoff_enabled,
+        require_surface_interest,
     );
 
     if plan == RemoteInstallRunningServerPlan::KeepRunning {
@@ -1172,37 +1194,57 @@ fn confirm_remote_install_with_running_server(
     Ok(true)
 }
 
-fn remote_install_running_server_plan(
-    endpoint_protocol_generation: Option<u32>,
-    detached_server_daemon: bool,
-    live_handoff: bool,
-    live_handoff_enabled: bool,
-) -> RemoteInstallRunningServerPlan {
-    let Some(reason) =
-        remote_server_restart_reason(endpoint_protocol_generation, detached_server_daemon)
-    else {
-        return RemoteInstallRunningServerPlan::KeepRunning;
-    };
-
-    if live_handoff_enabled && live_handoff {
-        return RemoteInstallRunningServerPlan::LiveHandoff;
-    }
-
-    RemoteInstallRunningServerPlan::StopRequired(reason)
-}
-
 fn remote_server_status(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
+    require_surface_interest: bool,
 ) -> io::Result<RemoteServerStatus> {
-    let command = format!("{} status server --json", remote_herdr.shell_path);
+    let command = remote_session_command(remote_herdr, &ssh.session_name, "status server --json");
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
         return Err(command_failed("remote server status failed", &output));
     }
 
     let stdout = String::from_utf8_lossy(&output.stdout);
-    parse_remote_server_status_json(stdout.trim())
+    let status = parse_remote_server_status_json(stdout.trim())?;
+    if require_surface_interest
+        && matches!(
+            status,
+            RemoteServerStatus::Running {
+                endpoint_protocol_generation: Some(
+                    crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION
+                ),
+                surface_interest: true,
+                health_check: true,
+                ..
+            }
+        )
+    {
+        // Older status helpers omit newer capabilities. Ask the live endpoint rather than
+        // assuming that the installed binary and the running daemon support the same features.
+        let negotiation = probe_remote_endpoint(ssh, remote_herdr)?;
+        return Ok(status.with_endpoint_negotiation(&negotiation));
+    }
+    Ok(status)
+}
+
+fn probe_remote_endpoint(
+    ssh: &RemoteSsh,
+    remote_herdr: &RemoteHerdr,
+) -> io::Result<crate::client::endpoint::EndpointNegotiation> {
+    let path = local_forward_socket_path(ssh.target(), &ssh.session_name);
+    let _bridge = SshStdioBridge::start(
+        ssh.target.clone(),
+        remote_herdr.clone(),
+        path.clone(),
+        ssh.session_name.clone(),
+        None,
+        true,
+    )?;
+    let mut stream = crate::ipc::connect_local_stream(&path)?;
+    // Use the saved client's noninteractive path. This metadata-only attachment never
+    // acquires a surface or sends pane input.
+    crate::client::probe_endpoint_negotiation(&mut stream)
 }
 
 #[derive(Debug, Deserialize)]
@@ -1213,6 +1255,27 @@ struct RemoteClientStatusJson {
     protocol: Option<u32>,
     #[serde(default)]
     endpoint_protocol_generation: Option<u32>,
+    #[serde(default)]
+    endpoint_capabilities: Vec<String>,
+}
+
+impl RemoteClientStatusJson {
+    fn supports_endpoint_requirement(&self, require_surface_interest: bool) -> bool {
+        self.endpoint_protocol_generation
+            == Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION)
+            && (!require_surface_interest
+                || [
+                    crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY,
+                    crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY,
+                    crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY,
+                ]
+                .iter()
+                .all(|required| {
+                    self.endpoint_capabilities
+                        .iter()
+                        .any(|capability| capability == required)
+                }))
+    }
 }
 
 #[derive(Debug, Deserialize)]
@@ -1229,6 +1292,10 @@ struct RemoteServerCapabilitiesJson {
     detached_server_daemon: bool,
     #[serde(default)]
     endpoint_protocol_generation: Option<u32>,
+    #[serde(default)]
+    surface_interest: bool,
+    #[serde(default)]
+    health_check: bool,
 }
 
 fn parse_client_status_json(status: &str) -> Option<RemoteClientStatusJson> {
@@ -1241,6 +1308,7 @@ fn parse_client_status_json(status: &str) -> Option<RemoteClientStatusJson> {
             status.version.is_some()
                 || status.protocol.is_some()
                 || status.endpoint_protocol_generation.is_some()
+                || !status.endpoint_capabilities.is_empty()
         })
 }
 
@@ -1261,6 +1329,12 @@ fn parse_remote_server_status_json(status: &str) -> io::Result<RemoteServerStatu
         endpoint_protocol_generation: capabilities
             .as_ref()
             .and_then(|capabilities| capabilities.endpoint_protocol_generation),
+        surface_interest: capabilities
+            .as_ref()
+            .is_some_and(|capabilities| capabilities.surface_interest),
+        health_check: capabilities
+            .as_ref()
+            .is_some_and(|capabilities| capabilities.health_check),
         live_handoff: capabilities
             .as_ref()
             .is_some_and(|capabilities| capabilities.live_handoff),
@@ -1275,8 +1349,14 @@ fn confirm_remote_server_stop(
     version: Option<&str>,
     reason: RemoteServerRestartReason,
 ) -> io::Result<bool> {
+    let required_upgrade = matches!(
+        reason,
+        RemoteServerRestartReason::EndpointProtocol
+            | RemoteServerRestartReason::SurfaceInterest
+            | RemoteServerRestartReason::HealthCheck
+    );
     if !io::stdin().is_terminal() {
-        if reason == RemoteServerRestartReason::EndpointProtocolMissing {
+        if required_upgrade {
             return Err(io::Error::other(format!(
                 "remote herdr server on {target} needs one final update before this client can attach; run from an interactive terminal to approve updating it"
             )));
@@ -1296,19 +1376,27 @@ fn confirm_remote_server_stop(
     eprintln!();
 
     match reason {
-        RemoteServerRestartReason::EndpointProtocolMissing => {
+        RemoteServerRestartReason::EndpointProtocol => {
             eprintln!(
                 "the remote server predates Herdr's stable endpoint protocol and must update before this client can attach."
             );
         }
-        RemoteServerRestartReason::DaemonDetachMissing => {
+        RemoteServerRestartReason::SurfaceInterest => {
+            eprintln!(
+                "the remote server must restart before it can join saved SSH endpoint federation."
+            );
+        }
+        RemoteServerRestartReason::HealthCheck => {
+            eprintln!("the remote server must restart to enable saved SSH endpoint health checks.");
+        }
+        RemoteServerRestartReason::DaemonDetach => {
             eprintln!(
                 "the remote server was started by a herdr build that may not survive SSH connection loss. restart it so network drops disconnect only this client."
             );
         }
     }
 
-    let prompt = if reason == RemoteServerRestartReason::EndpointProtocolMissing {
+    let prompt = if required_upgrade {
         "update the remote server and continue attaching? [Y/n] "
     } else {
         "restart the remote server now? [y/N] "
@@ -1316,16 +1404,10 @@ fn confirm_remote_server_stop(
     eprint!("{prompt}");
     io::stderr().flush()?;
 
-    let mut answer = String::new();
-    io::stdin().read_line(&mut answer)?;
-    let answer = answer.trim().to_ascii_lowercase();
-    if answer == "y" || answer == "yes" {
+    if read_remote_confirmation(&mut io::stdin().lock(), required_upgrade)? {
         return Ok(true);
     }
-    if answer.is_empty() && reason == RemoteServerRestartReason::EndpointProtocolMissing {
-        return Ok(true);
-    }
-    if reason == RemoteServerRestartReason::EndpointProtocolMissing {
+    if required_upgrade {
         return Err(io::Error::new(
             io::ErrorKind::Interrupted,
             "remote herdr server stop cancelled",
@@ -1335,10 +1417,19 @@ fn confirm_remote_server_stop(
     Ok(false)
 }
 
-fn remote_live_handoff_command(remote_herdr: &RemoteHerdr, protocol: u32, version: &str) -> String {
-    format!(
-        "{} server live-handoff --import-exe {} --expected-protocol {} --expected-version {}",
-        remote_herdr.shell_path, remote_herdr.shell_path, protocol, version
+fn remote_live_handoff_command(
+    remote_herdr: &RemoteHerdr,
+    session_name: &str,
+    protocol: u32,
+    version: &str,
+) -> String {
+    remote_session_command(
+        remote_herdr,
+        session_name,
+        &format!(
+            "server live-handoff --import-exe {} --expected-protocol {} --expected-version {}",
+            remote_herdr.shell_path, protocol, version
+        ),
     )
 }
 
@@ -1353,7 +1444,7 @@ fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io
         .version
         .filter(|version| !version.is_empty())
         .ok_or_else(|| io::Error::other("prepared remote herdr did not report its version"))?;
-    let command = remote_live_handoff_command(remote_herdr, protocol, &version);
+    let command = remote_live_handoff_command(remote_herdr, &ssh.session_name, protocol, &version);
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
         return Err(command_failed("remote server live handoff failed", &output));
@@ -1367,7 +1458,7 @@ fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io
 }
 
 fn stop_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
-    let command = format!("{} server stop", remote_herdr.shell_path);
+    let command = remote_session_command(remote_herdr, &ssh.session_name, "server stop");
     let output = ssh.sh_output(&command)?;
     if !output.status.success() {
         return Err(command_failed("remote server stop failed", &output));
@@ -1384,7 +1475,7 @@ fn stop_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result
 fn wait_for_remote_server_shutdown(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
     let deadline = Instant::now() + REMOTE_SERVER_SHUTDOWN_CONFIRM_TIMEOUT;
     loop {
-        if remote_server_status(ssh, remote_herdr)? == RemoteServerStatus::NotRunning {
+        if remote_server_status(ssh, remote_herdr, false)? == RemoteServerStatus::NotRunning {
             return Ok(());
         }
         if Instant::now() >= deadline {
@@ -1577,6 +1668,25 @@ fn private_download_dir(asset_key: &str) -> io::Result<PathBuf> {
     ))
 }
 
+fn read_remote_confirmation(reader: &mut impl io::BufRead, default: bool) -> io::Result<bool> {
+    let mut answer = String::new();
+    if reader.read_line(&mut answer)? == 0 {
+        return Err(io::Error::new(
+            io::ErrorKind::Interrupted,
+            "remote setup cancelled",
+        ));
+    }
+    match answer.trim().to_ascii_lowercase().as_str() {
+        "y" | "yes" => Ok(true),
+        "n" | "no" => Ok(false),
+        "" => Ok(default),
+        _ => Err(io::Error::new(
+            io::ErrorKind::Interrupted,
+            "remote setup cancelled: expected yes or no",
+        )),
+    }
+}
+
 fn confirm_remote_install(
     target: &str,
     remote_herdr: &RemoteHerdr,
@@ -1601,10 +1711,7 @@ fn confirm_remote_install(
     );
     io::stderr().flush()?;
 
-    let mut answer = String::new();
-    io::stdin().read_line(&mut answer)?;
-    let answer = answer.trim().to_ascii_lowercase();
-    if answer == "n" || answer == "no" {
+    if !read_remote_confirmation(&mut io::stdin().lock(), true)? {
         return Err(io::Error::new(
             io::ErrorKind::Interrupted,
             "remote herdr installation cancelled",
@@ -1614,16 +1721,24 @@ fn confirm_remote_install(
     Ok(())
 }
 
-fn remote_bridge_command(remote_herdr: &RemoteHerdr, session_name: &str) -> String {
-    let mut command = format!("exec {}", remote_herdr.shell_path);
+fn remote_session_command(remote_herdr: &RemoteHerdr, session_name: &str, args: &str) -> String {
+    let mut command = remote_herdr.shell_path.clone();
     if session_name != crate::session::DEFAULT_SESSION_NAME {
         command.push_str(" --session ");
         command.push_str(&shell_quote(session_name));
     }
-    command.push_str(" remote-client-bridge");
+    command.push(' ');
+    command.push_str(args);
     command
 }
 
+fn remote_bridge_command(remote_herdr: &RemoteHerdr, session_name: &str) -> String {
+    format!(
+        "exec {}",
+        remote_session_command(remote_herdr, session_name, "remote-client-bridge")
+    )
+}
+
 fn reattach_command(
     program: &str,
     target: &str,
@@ -1658,7 +1773,7 @@ fn command_failed(context: &str, output: &Output) -> io::Error {
     }
 }
 
-struct SshStdioBridge {
+pub(super) struct SshStdioBridge {
     local_socket: PathBuf,
     socket_identity: crate::ipc::SocketFileIdentity,
     should_stop: Arc<AtomicBool>,
@@ -1666,12 +1781,13 @@ struct SshStdioBridge {
 }
 
 impl SshStdioBridge {
-    fn start(
+    pub(super) fn start(
         target: String,
         remote_herdr: RemoteHerdr,
         local_socket: PathBuf,
         session_name: String,
         ssh_options: Option<&ManagedSshOptions>,
+        noninteractive: bool,
     ) -> io::Result<Self> {
         crate::ipc::prepare_socket_path(&local_socket, |path| {
             format!("remote bridge is already listening at {}", path.display())
@@ -1712,16 +1828,25 @@ impl SshStdioBridge {
                             &remote_herdr,
                             &session_name,
                             thread_ssh_options.as_ref(),
+                            noninteractive,
                             &thread_stop,
                         ) {
-                            eprintln!("herdr: remote bridge failed: {err}");
+                            if noninteractive {
+                                tracing::warn!(error = %err, "saved SSH endpoint bridge failed");
+                            } else {
+                                eprintln!("herdr: remote bridge failed: {err}");
+                            }
                         }
                     }
                     Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
                         thread::sleep(BRIDGE_ACCEPT_POLL);
                     }
                     Err(err) => {
-                        eprintln!("herdr: remote bridge listener failed: {err}");
+                        if noninteractive {
+                            tracing::warn!(error = %err, "saved SSH endpoint listener failed");
+                        } else {
+                            eprintln!("herdr: remote bridge listener failed: {err}");
+                        }
                         break;
                     }
                 }
@@ -1813,79 +1938,31 @@ fn write_managed_ssh_config() -> io::Result<ManagedSshConfig> {
     })
 }
 
-#[cfg(unix)]
-fn bridge_connection(
-    stream: crate::ipc::LocalStream,
-    target: &str,
-    remote_herdr: &RemoteHerdr,
-    session_name: &str,
-    ssh_options: Option<&ManagedSshOptions>,
-    _bridge_stop: &Arc<AtomicBool>,
-) -> io::Result<()> {
-    let mut command = Command::new("ssh");
-    apply_managed_ssh_options(&mut command, ssh_options);
-    command
-        .arg("-T")
-        .arg(target)
-        .arg(remote_bridge_command(remote_herdr, session_name))
-        .stdin(Stdio::piped())
-        .stdout(Stdio::piped())
-        .stderr(Stdio::inherit());
-
-    let mut child = command
-        .spawn()
-        .map_err(|err| io::Error::new(err.kind(), format!("failed to start ssh bridge: {err}")))?;
-    let mut child_stdin = child
-        .stdin
-        .take()
-        .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "ssh bridge stdin missing"))?;
-    let mut child_stdout = child
-        .stdout
-        .take()
-        .ok_or_else(|| io::Error::new(io::ErrorKind::BrokenPipe, "ssh bridge stdout missing"))?;
-    let mut stream_to_child = stream.try_clone()?;
-    let mut child_to_stream = stream;
-
-    let upload = thread::spawn(move || {
-        let _ = copy_flush(&mut stream_to_child, &mut child_stdin);
-    });
-    let download = thread::spawn(move || {
-        let _ = copy_flush(&mut child_stdout, &mut child_to_stream);
-        let _ = crate::ipc::shutdown_local_stream_write(&child_to_stream);
-    });
-
-    let status = child.wait()?;
-    let _ = upload.join();
-    let _ = download.join();
-
-    if status.success() {
-        Ok(())
-    } else {
-        Err(io::Error::new(
-            io::ErrorKind::ConnectionAborted,
-            format!("ssh bridge exited with {status}"),
-        ))
-    }
-}
-
-#[cfg(windows)]
 fn bridge_connection(
     stream: crate::ipc::LocalStream,
     target: &str,
     remote_herdr: &RemoteHerdr,
     session_name: &str,
     ssh_options: Option<&ManagedSshOptions>,
+    noninteractive: bool,
     bridge_stop: &Arc<AtomicBool>,
 ) -> io::Result<()> {
     let mut command = Command::new("ssh");
     apply_managed_ssh_options(&mut command, ssh_options);
+    if noninteractive {
+        apply_noninteractive_ssh_options(&mut command);
+    }
     command
         .arg("-T")
         .arg(target)
         .arg(remote_bridge_command(remote_herdr, session_name))
         .stdin(Stdio::piped())
         .stdout(Stdio::piped())
-        .stderr(Stdio::inherit());
+        .stderr(if noninteractive {
+            Stdio::null()
+        } else {
+            Stdio::inherit()
+        });
 
     let mut child = command
         .spawn()
@@ -2018,33 +2095,12 @@ fn bridge_connection(
     }
 }
 
-#[cfg(unix)]
-fn copy_flush<R: io::Read, W: io::Write>(reader: &mut R, writer: &mut W) -> io::Result<u64> {
-    let mut buffer = [0_u8; 16 * 1024];
-    let mut total = 0;
-
-    loop {
-        let bytes_read = match reader.read(&mut buffer) {
-            Ok(0) => return Ok(total),
-            Ok(bytes_read) => bytes_read,
-            Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
-            Err(err) => return Err(err),
-        };
-
-        writer.write_all(&buffer[..bytes_read])?;
-        writer.flush()?;
-        total += bytes_read as u64;
-    }
-}
-
-#[cfg(windows)]
 fn terminate_bridge_child(mut child: std::process::Child, message: &'static str) -> io::Result<()> {
     let _ = child.kill();
     let _ = child.wait();
     Err(io::Error::new(io::ErrorKind::BrokenPipe, message))
 }
 
-#[cfg(windows)]
 fn copy_reader_to_local_stream<R: io::Read>(
     reader: &mut R,
     stream: &mut crate::ipc::LocalStream,
@@ -2082,7 +2138,6 @@ fn copy_reader_to_local_stream<R: io::Read>(
     }
 }
 
-#[cfg(windows)]
 fn copy_local_stream_to_writer<W: io::Write>(
     mut stream: crate::ipc::LocalStream,
     writer: &mut W,
@@ -2207,6 +2262,7 @@ mod tests {
             socket.clone(),
             "default".to_string(),
             None,
+            false,
         )
         .expect("start bridge listener");
 
@@ -2253,9 +2309,8 @@ mod tests {
         let _ = std::fs::remove_file(socket);
     }
 
-    #[cfg(windows)]
     #[test]
-    fn windows_bridge_drop_while_waiting_for_client_is_bounded() {
+    fn bridge_drop_while_waiting_for_client_is_bounded() {
         let socket = local_forward_socket_path("drop-test", "default");
         let remote_herdr = RemoteHerdr::for_platform(RemotePlatform {
             os: "linux",
@@ -2267,6 +2322,7 @@ mod tests {
             socket.clone(),
             "default".to_string(),
             None,
+            false,
         )
         .expect("start bridge listener");
         let started = Instant::now();
@@ -2362,7 +2418,9 @@ mod tests {
             .expect("Unix managed config has a control path");
         let ssh = RemoteSsh {
             target: "example".to_string(),
+            session_name: crate::session::DEFAULT_SESSION_NAME.into(),
             managed_config: Some(managed_config),
+            noninteractive: false,
         };
 
         let command = ssh.command();
@@ -2400,7 +2458,9 @@ mod tests {
 
         let ssh = RemoteSsh {
             target: "example".to_string(),
+            session_name: crate::session::DEFAULT_SESSION_NAME.into(),
             managed_config: Some(managed_config),
+            noninteractive: false,
         };
         let args = ssh
             .command()
@@ -2428,10 +2488,106 @@ mod tests {
     }
 
     #[test]
+    fn noninteractive_ssh_command_cannot_prompt_or_accept_unknown_hosts() {
+        let ssh = RemoteSsh::new_noninteractive("example".into());
+        let args = ssh
+            .command()
+            .get_args()
+            .map(|arg| arg.to_string_lossy().into_owned())
+            .collect::<Vec<_>>();
+        for required in [
+            "BatchMode=yes",
+            "NumberOfPasswordPrompts=0",
+            "StrictHostKeyChecking=yes",
+            "ConnectTimeout=10",
+            "ConnectionAttempts=1",
+            "ServerAliveInterval=15",
+            "ServerAliveCountMax=4",
+        ] {
+            assert!(args.iter().any(|arg| arg == required), "missing {required}");
+        }
+        assert!(!args.iter().any(|arg| arg == "-F"));
+        assert!(ssh.options().is_none());
+    }
+
+    #[test]
+    fn remote_setup_approval_requires_input_and_rejects_unrecognized_answers() {
+        for default in [false, true] {
+            for input in ["", "maybe\n"] {
+                assert_eq!(
+                    read_remote_confirmation(&mut input.as_bytes(), default)
+                        .unwrap_err()
+                        .kind(),
+                    io::ErrorKind::Interrupted
+                );
+            }
+            assert_eq!(
+                read_remote_confirmation(&mut "\n".as_bytes(), default).unwrap(),
+                default
+            );
+            assert!(read_remote_confirmation(&mut "YES\n".as_bytes(), default).unwrap());
+            assert!(!read_remote_confirmation(&mut "no\n".as_bytes(), default).unwrap());
+        }
+    }
+
+    #[test]
+    fn saved_machine_compatibility_uses_capabilities_not_release_or_private_protocol() {
+        let mut status = RemoteClientStatusJson {
+            version: Some("0.1.0".into()),
+            protocol: Some(1),
+            endpoint_protocol_generation: Some(
+                crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION,
+            ),
+            endpoint_capabilities: vec![
+                crate::protocol::endpoint::SURFACE_INTEREST_CAPABILITY.into(),
+                crate::protocol::endpoint::PRESENTATION_EFFECTS_FENCE_CAPABILITY.into(),
+                crate::protocol::endpoint::HEALTH_CHECK_CAPABILITY.into(),
+            ],
+        };
+        assert!(status.supports_endpoint_requirement(true));
+        for index in 0..status.endpoint_capabilities.len() {
+            let removed = status.endpoint_capabilities.remove(index);
+            assert!(!status.supports_endpoint_requirement(true));
+            assert!(status.supports_endpoint_requirement(false));
+            status.endpoint_capabilities.insert(index, removed);
+        }
+        status.endpoint_protocol_generation = None;
+        assert!(!status.supports_endpoint_requirement(true));
+    }
+
+    #[test]
+    fn saved_machine_server_commands_are_scoped_to_the_explicit_session() {
+        let herdr =
+            RemoteHerdr::for_platform(RemotePlatform::from_uname("Linux", "x86_64").unwrap());
+        for command in [
+            "status server --json",
+            "server stop",
+            "remote-client-bridge",
+        ] {
+            assert_eq!(
+                remote_session_command(&herdr, "agents", command),
+                format!("{} --session agents {command}", herdr.shell_path)
+            );
+            assert_eq!(
+                remote_session_command(&herdr, crate::session::DEFAULT_SESSION_NAME, command),
+                format!("{} {command}", herdr.shell_path)
+            );
+        }
+        assert!(
+            remote_live_handoff_command(&herdr, "agents", 19, "0.7.9").starts_with(&format!(
+                "{} --session agents server live-handoff",
+                herdr.shell_path
+            ))
+        );
+    }
+
+    #[test]
     fn remote_ssh_command_is_plain_without_managed_config() {
         let ssh = RemoteSsh {
             target: "example".to_string(),
+            session_name: crate::session::DEFAULT_SESSION_NAME.into(),
             managed_config: None,
+            noninteractive: false,
         };
 
         let command = ssh.command();
@@ -2898,27 +3054,91 @@ mod tests {
     #[test]
     fn parse_client_status_json_reads_last_json_record() {
         let status = parse_client_status_json(
-            "wrapper output\n{\"version\":\"0.8.0\",\"protocol\":20,\"endpoint_protocol_generation\":1}\n{\"wrapper\":true}\n",
+            "wrapper output\n{\"version\":\"0.8.0\",\"protocol\":20,\"endpoint_protocol_generation\":1,\"endpoint_capabilities\":[\"surface_interest\",\"health_check\"]}\n{\"wrapper\":true}\n",
         )
         .unwrap();
         assert_eq!(status.version.as_deref(), Some("0.8.0"));
         assert_eq!(status.protocol, Some(20));
         assert_eq!(status.endpoint_protocol_generation, Some(1));
+        assert_eq!(
+            status.endpoint_capabilities,
+            vec!["surface_interest", "health_check"]
+        );
         assert!(
             parse_client_status_json(r#"{"endpoint_protocol_generation":"unknown"}"#).is_none()
         );
     }
 
     #[test]
+    fn saved_machine_setup_handoffs_old_server_missing_presentation_fence() {
+        // Captured from Rohan after installing a new binary while the old daemon stayed alive.
+        let installed = parse_client_status_json(
+            r#"{"version":"0.8.2","protocol":22,"endpoint_protocol_generation":1,"endpoint_capabilities":["surface_interest","presentation_effects_fence","health_check"]}"#,
+        )
+        .unwrap();
+        let running_binary = parse_client_status_json(
+            r#"{"version":"0.8.2","protocol":22,"endpoint_protocol_generation":1,"endpoint_capabilities":["surface_interest","health_check"]}"#,
+        )
+        .unwrap();
+        assert!(installed.supports_endpoint_requirement(true));
+        assert!(!running_binary.supports_endpoint_requirement(true));
+        for (live_capabilities, expected) in [
+            (
+                running_binary.endpoint_capabilities,
+                RemoteInstallRunningServerPlan::LiveHandoff,
+            ),
+            (
+                installed.endpoint_capabilities,
+                RemoteInstallRunningServerPlan::KeepRunning,
+            ),
+        ] {
+            let live_negotiation = crate::client::endpoint::EndpointNegotiation::new(
+                vec!["client_shell.surface.set".into()],
+                live_capabilities,
+            );
+            let RemoteServerStatus::Running {
+                endpoint_protocol_generation,
+                surface_interest,
+                health_check,
+                live_handoff,
+                detached_server_daemon,
+                ..
+            } = parse_remote_server_status_json(
+                r#"{"status":"running","running":true,"version":"0.8.2","protocol":22,"capabilities":{"live_handoff":true,"detached_server_daemon":true,"endpoint_protocol_generation":1,"surface_interest":true,"health_check":true}}"#,
+            )
+            .unwrap()
+            .with_endpoint_negotiation(&live_negotiation) else {
+                panic!("captured server must be running");
+            };
+
+            assert_eq!(
+                remote_install_running_server_plan(
+                    endpoint_protocol_generation,
+                    detached_server_daemon,
+                    surface_interest,
+                    health_check,
+                    live_handoff,
+                    true,
+                    true,
+                ),
+                expected,
+                "setup must follow the running server's negotiated capabilities",
+            );
+        }
+    }
+
+    #[test]
     fn parse_remote_server_status_json_reads_running_server() {
         assert_eq!(
             parse_remote_server_status_json(
-                r#"{"status":"running","running":true,"version":"0.6.0","protocol":8,"capabilities":{"live_handoff":true,"detached_server_daemon":true,"endpoint_protocol_generation":1}}"#
+                r#"{"status":"running","running":true,"version":"0.6.0","protocol":8,"capabilities":{"live_handoff":true,"detached_server_daemon":true,"endpoint_protocol_generation":1,"surface_interest":true,"health_check":true}}"#
             )
             .unwrap(),
             RemoteServerStatus::Running {
                 version: Some("0.6.0".into()),
                 endpoint_protocol_generation: Some(1),
+                surface_interest: true,
+                health_check: true,
                 live_handoff: true,
                 detached_server_daemon: true
             }
@@ -2935,6 +3155,8 @@ mod tests {
             RemoteServerStatus::Running {
                 version: Some("0.6.0".into()),
                 endpoint_protocol_generation: None,
+                surface_interest: false,
+                health_check: false,
                 live_handoff: false,
                 detached_server_daemon: false
             }
@@ -3109,88 +3331,17 @@ mod tests {
     }
 
     #[test]
-    fn remote_server_restart_reason_requires_one_update_for_pre_floor_server() {
-        assert_eq!(
-            remote_server_restart_reason(None, true),
-            Some(RemoteServerRestartReason::EndpointProtocolMissing)
-        );
-    }
-
-    #[test]
-    fn remote_server_restart_reason_allows_compatible_server() {
-        assert_eq!(
-            remote_server_restart_reason(
-                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
-                true
-            ),
-            None
-        );
-    }
-
-    #[test]
-    fn remote_server_restart_reason_requires_restart_for_old_daemon() {
-        assert_eq!(
-            remote_server_restart_reason(
-                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
-                false
-            ),
-            Some(RemoteServerRestartReason::DaemonDetachMissing)
-        );
-    }
-
-    #[test]
-    fn remote_install_plan_keeps_compatible_running_server() {
-        assert_eq!(
-            remote_install_running_server_plan(
-                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
-                true,
-                false,
-                false
-            ),
-            RemoteInstallRunningServerPlan::KeepRunning
-        );
-    }
-
-    #[test]
-    fn remote_install_plan_requires_stop_for_old_daemon() {
-        assert_eq!(
-            remote_install_running_server_plan(
-                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
-                false,
-                false,
-                false
-            ),
-            RemoteInstallRunningServerPlan::StopRequired(
-                RemoteServerRestartReason::DaemonDetachMissing
-            )
-        );
-    }
-
-    #[test]
-    fn remote_install_plan_requires_stop_for_pre_floor_server() {
-        assert_eq!(
-            remote_install_running_server_plan(None, true, false, false),
-            RemoteInstallRunningServerPlan::StopRequired(
-                RemoteServerRestartReason::EndpointProtocolMissing
-            )
-        );
-    }
-
-    #[test]
-    fn remote_install_plan_uses_live_handoff_for_pre_floor_server() {
-        assert_eq!(
-            remote_install_running_server_plan(None, true, true, true),
-            RemoteInstallRunningServerPlan::LiveHandoff
-        );
-    }
-
-    #[test]
     fn remote_live_handoff_uses_prepared_binary_identity() {
         let remote_herdr = RemoteHerdr::for_platform(RemotePlatform {
             os: "linux",
             arch: "x86_64",
         });
-        let command = remote_live_handoff_command(&remote_herdr, 19, "0.7.9");
+        let command = remote_live_handoff_command(
+            &remote_herdr,
+            crate::session::DEFAULT_SESSION_NAME,
+            19,
+            "0.7.9",
+        );
         assert!(command.contains("--expected-protocol 19"));
         assert!(command.contains("--expected-version 0.7.9"));
         assert!(!command.contains(&format!(
diff --git a/src/remote/process.rs b/src/remote/process.rs
new file mode 100644
index 00000000..d3db5c9f
--- /dev/null
+++ b/src/remote/process.rs
@@ -0,0 +1,87 @@
+use std::io::{self, Read as _};
+use std::process::Output;
+use std::thread;
+use std::time::{Duration, Instant};
+
+const POLL_INTERVAL: Duration = Duration::from_millis(50);
+
+pub(super) fn wait_with_output_timeout(
+    mut child: std::process::Child,
+    timeout: Duration,
+) -> io::Result<Output> {
+    let stdout = child
+        .stdout
+        .take()
+        .ok_or_else(|| io::Error::other("SSH command stdout was not captured"))?;
+    let stderr = child
+        .stderr
+        .take()
+        .ok_or_else(|| io::Error::other("SSH command stderr was not captured"))?;
+    let stdout = thread::spawn(move || {
+        let mut stdout = stdout;
+        let mut bytes = Vec::new();
+        stdout.read_to_end(&mut bytes).map(|_| bytes)
+    });
+    let stderr = thread::spawn(move || {
+        let mut stderr = stderr;
+        let mut bytes = Vec::new();
+        stderr.read_to_end(&mut bytes).map(|_| bytes)
+    });
+    let started = Instant::now();
+    let status = loop {
+        match child.try_wait() {
+            Ok(Some(status)) => break status,
+            Ok(None) => {}
+            Err(error) => {
+                let _ = child.kill();
+                let _ = child.wait();
+                let _ = stdout.join();
+                let _ = stderr.join();
+                return Err(error);
+            }
+        }
+        if started.elapsed() >= timeout {
+            let _ = child.kill();
+            let _ = child.wait();
+            let _ = stdout.join();
+            let _ = stderr.join();
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "noninteractive SSH command timed out",
+            ));
+        }
+        thread::sleep(POLL_INTERVAL);
+    };
+    let stdout = stdout
+        .join()
+        .map_err(|_| io::Error::other("SSH stdout reader panicked"))??;
+    let stderr = stderr
+        .join()
+        .map_err(|_| io::Error::other("SSH stderr reader panicked"))??;
+    Ok(Output {
+        status,
+        stdout,
+        stderr,
+    })
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::process::{Command, Stdio};
+
+    #[test]
+    fn timeout_kills_the_child() {
+        let mut command = Command::new("sh");
+        command
+            .arg("-c")
+            .arg("exec sleep 10")
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped());
+        let started = Instant::now();
+        let error = wait_with_output_timeout(command.spawn().unwrap(), Duration::from_millis(25))
+            .unwrap_err();
+        assert_eq!(error.kind(), io::ErrorKind::TimedOut);
+        assert!(started.elapsed() < Duration::from_secs(1));
+    }
+}
diff --git a/src/remote/restart_policy.rs b/src/remote/restart_policy.rs
new file mode 100644
index 00000000..757efcc5
--- /dev/null
+++ b/src/remote/restart_policy.rs
@@ -0,0 +1,161 @@
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum RemoteServerRestartReason {
+    EndpointProtocol,
+    SurfaceInterest,
+    HealthCheck,
+    DaemonDetach,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum RemoteInstallRunningServerPlan {
+    KeepRunning,
+    LiveHandoff,
+    StopRequired(RemoteServerRestartReason),
+}
+
+pub(super) fn remote_server_restart_reason(
+    endpoint_protocol_generation: Option<u32>,
+    detached_server_daemon: bool,
+    require_surface_interest: bool,
+    surface_interest: bool,
+    health_check: bool,
+) -> Option<RemoteServerRestartReason> {
+    if endpoint_protocol_generation != Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION)
+    {
+        return Some(RemoteServerRestartReason::EndpointProtocol);
+    }
+    if require_surface_interest && !surface_interest {
+        return Some(RemoteServerRestartReason::SurfaceInterest);
+    }
+    if require_surface_interest && !health_check {
+        return Some(RemoteServerRestartReason::HealthCheck);
+    }
+    if !detached_server_daemon {
+        return Some(RemoteServerRestartReason::DaemonDetach);
+    }
+    None
+}
+
+pub(super) fn remote_install_running_server_plan(
+    endpoint_protocol_generation: Option<u32>,
+    detached_server_daemon: bool,
+    surface_interest: bool,
+    health_check: bool,
+    live_handoff: bool,
+    live_handoff_enabled: bool,
+    require_surface_interest: bool,
+) -> RemoteInstallRunningServerPlan {
+    let Some(reason) = remote_server_restart_reason(
+        endpoint_protocol_generation,
+        detached_server_daemon,
+        require_surface_interest,
+        surface_interest,
+        health_check,
+    ) else {
+        return RemoteInstallRunningServerPlan::KeepRunning;
+    };
+
+    if live_handoff_enabled && live_handoff {
+        return RemoteInstallRunningServerPlan::LiveHandoff;
+    }
+    RemoteInstallRunningServerPlan::StopRequired(reason)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn pre_floor_server_requires_update() {
+        assert_eq!(
+            remote_server_restart_reason(None, true, false, false, false),
+            Some(RemoteServerRestartReason::EndpointProtocol)
+        );
+    }
+
+    #[test]
+    fn compatible_server_can_keep_running() {
+        assert_eq!(
+            remote_server_restart_reason(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true,
+                false,
+                false,
+                false
+            ),
+            None
+        );
+    }
+
+    #[test]
+    fn saved_endpoint_requires_surface_interest() {
+        assert_eq!(
+            remote_server_restart_reason(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true,
+                true,
+                false,
+                false
+            ),
+            Some(RemoteServerRestartReason::SurfaceInterest)
+        );
+    }
+
+    #[test]
+    fn saved_endpoint_requires_health_checks() {
+        assert_eq!(
+            remote_server_restart_reason(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true,
+                true,
+                true,
+                false
+            ),
+            Some(RemoteServerRestartReason::HealthCheck)
+        );
+    }
+
+    #[test]
+    fn saved_endpoint_accepts_complete_lifecycle_support() {
+        assert_eq!(
+            remote_server_restart_reason(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true,
+                true,
+                true,
+                true
+            ),
+            None
+        );
+    }
+
+    #[test]
+    fn old_daemon_requires_restart() {
+        assert_eq!(
+            remote_server_restart_reason(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                false,
+                false,
+                false,
+                false
+            ),
+            Some(RemoteServerRestartReason::DaemonDetach)
+        );
+    }
+
+    #[test]
+    fn saved_endpoint_can_explicitly_live_handoff_for_surface_support() {
+        assert_eq!(
+            remote_install_running_server_plan(
+                Some(crate::protocol::endpoint::ENDPOINT_PROTOCOL_GENERATION),
+                true,
+                false,
+                false,
+                true,
+                true,
+                true
+            ),
+            RemoteInstallRunningServerPlan::LiveHandoff
+        );
+    }
+}
diff --git a/src/remote/saved.rs b/src/remote/saved.rs
new file mode 100644
index 00000000..47684758
--- /dev/null
+++ b/src/remote/saved.rs
@@ -0,0 +1,136 @@
+use std::io;
+use std::path::PathBuf;
+
+use super::attach::{find_installed_remote_herdr, RemoteSsh, SshStdioBridge};
+
+pub(crate) struct SavedSshBridge {
+    _bridge: SshStdioBridge,
+}
+
+pub(crate) struct SavedSshStream {
+    pub(crate) stream: crate::ipc::LocalStream,
+    pub(crate) bridge: SavedSshBridge,
+}
+
+pub(crate) fn connect_saved_ssh(
+    profile_id: &str,
+    target: &str,
+    session: &str,
+) -> io::Result<SavedSshStream> {
+    validate_profile_path_id(profile_id)?;
+    crate::session::validate_name(session)
+        .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
+    let ssh = RemoteSsh::new_noninteractive(target.to_owned());
+    let remote_herdr = find_installed_remote_herdr(&ssh)?;
+    let path = saved_bridge_path(profile_id);
+    let bridge = SshStdioBridge::start(
+        target.to_owned(),
+        remote_herdr,
+        path.clone(),
+        session.to_owned(),
+        ssh.options(),
+        true,
+    )?;
+    let stream = crate::ipc::connect_local_stream(&path)?;
+    Ok(SavedSshStream {
+        stream,
+        bridge: SavedSshBridge { _bridge: bridge },
+    })
+}
+
+pub(crate) fn saved_ssh_bootstrap_command(target: &str, session: &str) -> String {
+    format!(
+        "herdr --remote {} --session {} --handoff",
+        super::shell_quote(target),
+        super::shell_quote(session)
+    )
+}
+
+pub(crate) fn saved_ssh_failure_needs_attention(error: &io::Error) -> bool {
+    if matches!(
+        error.kind(),
+        io::ErrorKind::InvalidInput
+            | io::ErrorKind::InvalidData
+            | io::ErrorKind::NotFound
+            | io::ErrorKind::PermissionDenied
+            | io::ErrorKind::Unsupported
+    ) {
+        return true;
+    }
+    let message = error.to_string().to_ascii_lowercase();
+    [
+        "permission denied",
+        "host key verification failed",
+        "remote host identification has changed",
+        "no matching host key",
+        "unsupported remote platform",
+        "not ready",
+        "install or update",
+        "protocol",
+        "handshake",
+    ]
+    .iter()
+    .any(|needle| message.contains(needle))
+}
+
+fn saved_bridge_path(profile_id: &str) -> PathBuf {
+    let pid = std::process::id();
+    let readable = format!("herdr-ssh-{pid}-{profile_id}.sock");
+    let short = format!("herdr-s-{pid}-{}.sock", &profile_id[..16]);
+    crate::platform::remote_bridge_endpoint_path(&readable, &short)
+}
+
+fn validate_profile_path_id(profile_id: &str) -> io::Result<()> {
+    if profile_id.len() == 32
+        && profile_id
+            .bytes()
+            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
+    {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            "invalid SSH endpoint profile id",
+        ))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn bridge_paths_use_profile_identity_not_target_or_session() {
+        let first = saved_bridge_path("0123456789abcdef0123456789abcdef");
+        let second = saved_bridge_path("fedcba9876543210fedcba9876543210");
+        assert_ne!(first, second);
+        assert!(!first.to_string_lossy().contains("example.com"));
+        assert!(!first.to_string_lossy().contains("default"));
+    }
+
+    #[test]
+    fn bootstrap_command_preserves_the_explicit_remote_session() {
+        assert_eq!(
+            saved_ssh_bootstrap_command("build host", "agent work"),
+            "herdr --remote 'build host' --session 'agent work' --handoff"
+        );
+    }
+
+    #[test]
+    fn prompt_and_compatibility_failures_require_attention() {
+        for message in [
+            "Permission denied (publickey)",
+            "Host key verification failed",
+            "matching Herdr is not ready; install or update",
+            "handshake rejected",
+        ] {
+            assert!(saved_ssh_failure_needs_attention(&io::Error::other(
+                message
+            )));
+        }
+        assert!(!saved_ssh_failure_needs_attention(&io::Error::new(
+            io::ErrorKind::TimedOut,
+            "network timed out"
+        )));
+    }
+}