fix: cache saved machine connection metadata (#4256)

refs #4252

10 files changed, 611 insertions, 31 deletions

diff --git a/src/api/client.rs b/src/api/client.rs
index 326ebc7d..15d3a884 100644
--- a/src/api/client.rs
+++ b/src/api/client.rs
@@ -1,7 +1,7 @@
 use std::fmt;
-use std::io::{self, BufRead, BufReader, Write};
+use std::io::{self, BufRead, BufReader, Read, Write};
 use std::path::PathBuf;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use interprocess::local_socket::traits::Stream as _;
 use serde::de::DeserializeOwned;
@@ -75,10 +75,37 @@ impl ApiClient {
     }
 
     pub fn status(&self) -> Result<crate::api::RuntimeStatus, ApiClientError> {
-        let response = self.request(Request {
+        self.read_status(None)
+    }
+
+    pub(crate) fn status_with_timeout(
+        &self,
+        timeout: Duration,
+    ) -> Result<crate::api::RuntimeStatus, ApiClientError> {
+        self.read_status(Some(timeout))
+    }
+
+    fn read_status(
+        &self,
+        timeout: Option<Duration>,
+    ) -> Result<crate::api::RuntimeStatus, ApiClientError> {
+        let request = Request {
             id: "api-client:status".into(),
             method: Method::Ping(PingParams::default()),
-        })?;
+        };
+        let response = match timeout {
+            Some(timeout) => {
+                let mut stream = self.connect()?;
+                write_request(&mut stream, &request)?;
+                crate::ipc::set_local_stream_polling(&mut stream, true)?;
+                let mut reader = BufReader::new(DeadlineReader {
+                    stream: &mut stream,
+                    deadline: Instant::now() + timeout,
+                });
+                parse_response_value(read_json_line(&mut reader)?)?
+            }
+            None => self.request(request)?,
+        };
         match response.result {
             ResponseResult::Pong {
                 version,
@@ -162,9 +189,37 @@ fn write_request(stream: &mut LocalStream, request: &Request) -> Result<(), ApiC
     Ok(())
 }
 
-fn read_json_line<T: DeserializeOwned>(
-    reader: &mut BufReader<LocalStream>,
-) -> Result<T, ApiClientError> {
+struct DeadlineReader<'a> {
+    stream: &'a mut LocalStream,
+    deadline: Instant,
+}
+
+impl Read for DeadlineReader<'_> {
+    fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
+        if buffer.is_empty() {
+            return Ok(0);
+        }
+        loop {
+            if Instant::now() >= self.deadline {
+                return Err(io::Error::new(
+                    io::ErrorKind::TimedOut,
+                    "server status probe timed out",
+                ));
+            }
+            // Windows named pipes have no read timeout; peek-before-read keeps
+            // both idle and partial responses subject to the same deadline.
+            match crate::ipc::poll_local_stream_read_count(self.stream, buffer)? {
+                crate::ipc::LocalStreamReadCount::Data(count) => return Ok(count),
+                crate::ipc::LocalStreamReadCount::Closed => return Ok(0),
+                crate::ipc::LocalStreamReadCount::Pending => {
+                    std::thread::sleep(Duration::from_millis(2))
+                }
+            }
+        }
+    }
+}
+
+fn read_json_line<T: DeserializeOwned>(reader: &mut impl BufRead) -> Result<T, ApiClientError> {
     let mut line = String::new();
     let read = reader.read_line(&mut line)?;
     if read == 0 || line.trim().is_empty() {
@@ -200,6 +255,34 @@ mod tests {
     }
 
     #[test]
+    fn status_timeout_closes_a_stalled_probe() {
+        use interprocess::local_socket::traits::Listener as _;
+        let path =
+            std::env::temp_dir().join(format!("herdr-status-timeout-{}.sock", std::process::id()));
+        let listener = crate::ipc::bind_private_local_listener(&path).unwrap();
+        let server = std::thread::spawn(move || {
+            let stream = listener.accept().unwrap();
+            let mut reader = BufReader::new(stream);
+            let mut line = String::new();
+            reader.read_line(&mut line).unwrap();
+            assert_eq!(
+                serde_json::from_str::<serde_json::Value>(&line).unwrap()["method"],
+                "ping"
+            );
+            std::thread::sleep(Duration::from_millis(300));
+        });
+        let client = ApiClient::for_target(ConnectionTarget::SocketPath(path.clone()));
+        let error = client
+            .status_with_timeout(Duration::from_millis(100))
+            .unwrap_err();
+        assert!(
+            matches!(error, ApiClientError::Io(error) if matches!(error.kind(), io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock))
+        );
+        server.join().unwrap();
+        std::fs::remove_file(path).unwrap();
+    }
+
+    #[test]
     fn socket_path_target_uses_explicit_path() {
         let path = PathBuf::from("/tmp/herdr-test.sock");
         let client = ApiClient::for_target(ConnectionTarget::SocketPath(path.clone()));
diff --git a/src/cli.rs b/src/cli.rs
index 154e0c99..fead33b6 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -782,8 +782,7 @@ pub(super) fn send_request_unchecked(request: &Request) -> std::io::Result<serde
 }
 
 fn ensure_server_protocol_compatible(client: &ApiClient, request_id: &str) -> std::io::Result<()> {
-    let status = client
-        .status()
+    let status = target::server_status(client)
         .map_err(|err| map_server_not_running_or_io(err, request_id, client))?;
     let server_protocol = status
         .protocol
diff --git a/src/cli/machine.rs b/src/cli/machine.rs
index ab84041e..53404c03 100644
--- a/src/cli/machine.rs
+++ b/src/cli/machine.rs
@@ -164,18 +164,21 @@ fn add(args: &[String]) -> std::io::Result<i32> {
             return Ok(2);
         }
     }
-    if let Err(error) = crate::remote::prepare_saved_ssh(&target, &session) {
-        eprintln!("error: {error}; machine was not saved");
-        crate::remote::print_saved_ssh_error_hint(&error, &target);
-        return Ok(1);
-    }
+    let metadata = match crate::remote::prepare_saved_ssh(&target, &session) {
+        Ok(metadata) => metadata,
+        Err(error) => {
+            eprintln!("error: {error}; machine was not saved");
+            crate::remote::print_saved_ssh_error_hint(&error, &target);
+            return Ok(1);
+        }
+    };
     // Setup can wait for human approval. Do not overwrite catalog edits made meanwhile.
     let mut catalog = load_catalog().map_err(|error| {
         std::io::Error::other(format!(
             "remote prepared, but machine was not saved: {error}"
         ))
     })?;
-    let id = match catalog.add_ssh(label, target, session) {
+    let id = match catalog.add_ssh(label, &target, &session) {
         Ok(id) => id,
         Err(error) => {
             eprintln!("error: {error}");
@@ -187,6 +190,10 @@ fn add(args: &[String]) -> std::io::Result<i32> {
             "remote prepared, but machine was not saved: {error}"
         ))
     })?;
+    if let Some(metadata) = metadata {
+        crate::client::endpoint::SshMetadataCache::new(id.as_str(), &target, &session)?
+            .store(&metadata);
+    }
     println!("Saved SSH machine {id}. Remote server is ready.");
     println!("Open Herdr clients connect automatically.");
     Ok(0)
@@ -232,11 +239,26 @@ fn remove(args: &[String]) -> std::io::Result<i32> {
     };
     let mut catalog = load_catalog()?;
     let previous_selection = catalog.selected_profile.clone();
+    let metadata_cache = catalog
+        .ssh
+        .iter()
+        .find(|profile| profile.id == id)
+        .map(|profile| {
+            crate::client::endpoint::SshMetadataCache::new(
+                id.as_str(),
+                &profile.target,
+                &profile.session,
+            )
+        })
+        .transpose()?;
     if !catalog.remove_ssh(&id) {
         eprintln!("machine profile {id} was not found");
         return Ok(1);
     }
     store_catalog(&catalog)?;
+    if let Some(cache) = metadata_cache {
+        cache.invalidate();
+    }
     if catalog.selected_profile != previous_selection {
         catalog.store_selection().map_err(std::io::Error::other)?;
     }
diff --git a/src/cli/status.rs b/src/cli/status.rs
index d9c0be79..d13be784 100644
--- a/src/cli/status.rs
+++ b/src/cli/status.rs
@@ -174,7 +174,7 @@ fn print_server_status_body(server: &ServerRuntimeStatus, indent: &str) {
 }
 
 fn read_server_runtime_status() -> std::io::Result<ServerRuntimeStatus> {
-    match super::target::api_client()?.status() {
+    match super::target::server_status(&super::target::api_client()?) {
         Ok(status) => Ok(ServerRuntimeStatus::Running {
             version: status.version,
             protocol: status.protocol,
diff --git a/src/cli/target.rs b/src/cli/target.rs
index 64c072f9..220410f9 100644
--- a/src/cli/target.rs
+++ b/src/cli/target.rs
@@ -79,6 +79,7 @@ pub(super) fn api_client() -> io::Result<ApiClient> {
                     target.profile.id.as_str(),
                     &target.profile.target,
                     &target.profile.session,
+                    true,
                 )
                 .map_err(|error| {
                     io::Error::new(
@@ -98,6 +99,51 @@ pub(super) fn api_client() -> io::Result<ApiClient> {
     })
 }
 
+pub(super) fn server_status(
+    client: &ApiClient,
+) -> Result<crate::api::RuntimeStatus, crate::api::client::ApiClientError> {
+    let probe = || {
+        if is_remote() {
+            client.status_with_timeout(std::time::Duration::from_secs(15))
+        } else {
+            client.status()
+        }
+    };
+    let error = match probe() {
+        Ok(status) => return Ok(status),
+        Err(error) => error,
+    };
+    // Only this read-only probe may rediscover and retry. Requests that follow
+    // the probe must never be replayed after an ambiguous SSH failure.
+    TARGET.with(|target| {
+        let mut target = target.borrow_mut();
+        let Some(target) = target.as_mut() else {
+            return Err(error);
+        };
+        let Some(bridge) = target.bridge.as_ref() else {
+            return Err(error);
+        };
+        let Some(failure) = bridge.reported_failure() else {
+            return Err(error);
+        };
+        if !bridge.used_cached_metadata
+            || !crate::remote::SavedSshApiBridge::stale_metadata_failure(&failure)
+        {
+            return Err(failure.into());
+        }
+        bridge.invalidate_metadata();
+        target.bridge.take();
+        target.bridge = Some(crate::remote::SavedSshApiBridge::start(
+            target.profile.id.as_str(),
+            &target.profile.target,
+            &target.profile.session,
+            false,
+        )?);
+        Ok(())
+    })?;
+    probe()
+}
+
 pub(super) fn remote_error(error: io::Error) -> io::Error {
     TARGET.with(|target| {
         let target = target.borrow();
diff --git a/src/client/endpoint.rs b/src/client/endpoint.rs
index 77aaae3b..f6ae4109 100644
--- a/src/client/endpoint.rs
+++ b/src/client/endpoint.rs
@@ -9,6 +9,7 @@ mod control;
 mod health;
 mod message_policy;
 mod registry;
+mod ssh_metadata;
 mod supervisor;
 mod writer;
 
@@ -17,6 +18,7 @@ pub(crate) use catalog::*;
 pub(crate) use control::*;
 pub(crate) use message_policy::*;
 pub(crate) use registry::*;
+pub(crate) use ssh_metadata::{SshMachineMetadata, SshMetadataCache};
 pub(crate) use supervisor::*;
 pub(crate) use writer::NativeEndpointTransport;
 
diff --git a/src/client/endpoint/catalog.rs b/src/client/endpoint/catalog.rs
index 23f26387..e65d1e9e 100644
--- a/src/client/endpoint/catalog.rs
+++ b/src/client/endpoint/catalog.rs
@@ -331,7 +331,11 @@ fn load_selection_from_path(path: &Path) -> Result<Option<EndpointSelection>, St
     Ok(Some(selection))
 }
 
-fn store_private_json(path: &Path, content: &[u8], description: &str) -> Result<(), String> {
+pub(super) fn store_private_json(
+    path: &Path,
+    content: &[u8],
+    description: &str,
+) -> Result<(), String> {
     if content.len() as u64 > MAX_CATALOG_BYTES {
         return Err(format!("{description} exceeds the storage limit"));
     }
diff --git a/src/client/endpoint/ssh_metadata.rs b/src/client/endpoint/ssh_metadata.rs
new file mode 100644
index 00000000..bc8dffd6
--- /dev/null
+++ b/src/client/endpoint/ssh_metadata.rs
@@ -0,0 +1,221 @@
+use std::io::{self, Read as _};
+use std::path::{Path, PathBuf};
+
+use serde::{Deserialize, Serialize};
+
+use super::ProfileId;
+
+const MAX_METADATA_BYTES: u64 = 16 * 1024;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub(crate) struct SshMachineMetadata {
+    pub(crate) os: String,
+    pub(crate) executable: String,
+}
+
+impl SshMachineMetadata {
+    pub(crate) fn is_valid(&self) -> bool {
+        let path = &self.executable;
+        if path.is_empty() || path.len() > 4096 || path.chars().any(char::is_control) {
+            return false;
+        }
+        match self.os.as_str() {
+            "linux" | "macos" => path.starts_with('/') && !path.ends_with("/mise/shims/herdr"),
+            "windows" => {
+                let bytes = path.as_bytes();
+                path.starts_with(r"\\")
+                    || (bytes.len() >= 3
+                        && bytes[0].is_ascii_alphabetic()
+                        && bytes[1] == b':'
+                        && matches!(bytes[2], b'\\' | b'/'))
+            }
+            _ => false,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize)]
+struct StoredMetadata {
+    version: u32,
+    target: String,
+    session: String,
+    metadata: SshMachineMetadata,
+}
+
+pub(crate) struct SshMetadataCache {
+    path: PathBuf,
+    target: String,
+    session: String,
+}
+
+impl SshMetadataCache {
+    pub(crate) fn new(profile_id: &str, target: &str, session: &str) -> io::Result<Self> {
+        let id = ProfileId::parse(profile_id)
+            .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
+        Ok(Self {
+            path: crate::config::state_dir()
+                .join("client/ssh-metadata")
+                .join(format!("{id}.json")),
+            target: target.to_owned(),
+            session: session.to_owned(),
+        })
+    }
+
+    pub(crate) fn load(&self) -> Option<SshMachineMetadata> {
+        load_metadata(&self.path, &self.target, &self.session)
+    }
+
+    pub(crate) fn store(&self, metadata: &SshMachineMetadata) {
+        if !metadata.is_valid() {
+            return;
+        }
+        let stored = StoredMetadata {
+            version: 1,
+            target: self.target.clone(),
+            session: self.session.clone(),
+            metadata: metadata.clone(),
+        };
+        let result = serde_json::to_vec(&stored)
+            .map_err(|error| error.to_string())
+            .and_then(|bytes| {
+                super::catalog::store_private_json(&self.path, &bytes, "SSH metadata")
+            });
+        if let Err(error) = result {
+            tracing::debug!(%error, "could not cache SSH machine metadata");
+        }
+    }
+
+    pub(crate) fn invalidate(&self) {
+        if let Err(error) = std::fs::remove_file(&self.path) {
+            if error.kind() != io::ErrorKind::NotFound {
+                tracing::debug!(%error, "could not invalidate SSH machine metadata");
+            }
+        }
+    }
+}
+
+fn load_metadata(path: &Path, target: &str, session: &str) -> Option<SshMachineMetadata> {
+    let file_type = std::fs::symlink_metadata(path).ok()?.file_type();
+    if !file_type.is_file() {
+        return None;
+    }
+    let mut bytes = Vec::new();
+    std::fs::File::open(path)
+        .ok()?
+        .take(MAX_METADATA_BYTES + 1)
+        .read_to_end(&mut bytes)
+        .ok()?;
+    if bytes.len() as u64 > MAX_METADATA_BYTES {
+        return None;
+    }
+    let stored: StoredMetadata = serde_json::from_slice(&bytes).ok()?;
+    (stored.version == 1
+        && stored.target == target
+        && stored.session == session
+        && stored.metadata.is_valid())
+    .then_some(stored.metadata)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn metadata_accepts_only_supported_platforms_and_absolute_paths() {
+        for (os, path, valid) in [
+            ("linux", "/home/a b/herdr", true),
+            ("macos", "/opt/homebrew/bin/herdr", true),
+            ("windows", r"C:\Program Files\herdr.exe", true),
+            ("windows", r"\\server\share\herdr.exe", true),
+            ("windows", "herdr.exe", false),
+            ("linux", "$HOME/.local/bin/herdr", false),
+            ("linux", "/home/user/.local/share/mise/shims/herdr", false),
+            ("linux", "/bin/herdr\nmalformed", false),
+            ("unknown", "/bin/herdr", false),
+        ] {
+            assert_eq!(
+                SshMachineMetadata {
+                    os: os.into(),
+                    executable: path.into()
+                }
+                .is_valid(),
+                valid,
+                "{os}: {path}"
+            );
+        }
+    }
+
+    #[test]
+    fn metadata_is_disposable_fingerprinted_and_independent_per_profile() {
+        let root = std::env::temp_dir().join(format!("herdr-ssh-metadata-{}", std::process::id()));
+        let first = SshMetadataCache {
+            path: root.join("first.json"),
+            target: "mac".into(),
+            session: "fleet".into(),
+        };
+        let second = SshMetadataCache {
+            path: root.join("second.json"),
+            target: "mac".into(),
+            session: "fleet".into(),
+        };
+        let metadata = SshMachineMetadata {
+            os: "macos".into(),
+            executable: "/some path/herdr".into(),
+        };
+        assert!(first.load().is_none());
+        first.store(&metadata);
+        second.store(&metadata);
+        assert_eq!(first.load(), Some(metadata.clone()));
+        assert!(load_metadata(&first.path, "different-host", "fleet").is_none());
+        assert!(load_metadata(&first.path, "mac", "different-session").is_none());
+        #[cfg(unix)]
+        {
+            use std::os::unix::fs::PermissionsExt as _;
+            assert_eq!(
+                std::fs::metadata(&first.path).unwrap().permissions().mode() & 0o777,
+                0o600
+            );
+        }
+        let mut stored: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&first.path).unwrap()).unwrap();
+        stored["future_field"] = true.into();
+        std::fs::write(&first.path, serde_json::to_vec(&stored).unwrap()).unwrap();
+        assert_eq!(first.load(), Some(metadata.clone()));
+        stored["version"] = 2.into();
+        std::fs::write(&first.path, serde_json::to_vec(&stored).unwrap()).unwrap();
+        assert!(first.load().is_none());
+        for bytes in [
+            b"broken".to_vec(),
+            vec![b' '; MAX_METADATA_BYTES as usize + 1],
+        ] {
+            std::fs::write(&first.path, bytes).unwrap();
+            assert!(first.load().is_none());
+        }
+        first.invalidate();
+        assert_eq!(second.load(), Some(metadata));
+        std::fs::remove_dir_all(root).unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn metadata_does_not_follow_symlinks() {
+        let root =
+            std::env::temp_dir().join(format!("herdr-ssh-metadata-link-{}", std::process::id()));
+        std::fs::create_dir_all(&root).unwrap();
+        let cache = SshMetadataCache {
+            path: root.join("cache.json"),
+            target: "mac".into(),
+            session: "fleet".into(),
+        };
+        let other = root.join("other");
+        std::fs::write(&other, "untouched").unwrap();
+        std::os::unix::fs::symlink(&other, &cache.path).unwrap();
+        assert!(cache.load().is_none());
+        cache.store(&SshMachineMetadata {
+            os: "macos".into(),
+            executable: "/bin/herdr".into(),
+        });
+        assert_eq!(std::fs::read_to_string(&other).unwrap(), "untouched");
+        std::fs::remove_dir_all(root).unwrap();
+    }
+}
diff --git a/src/remote/attach.rs b/src/remote/attach.rs
index 6a1ee6f9..53023b83 100644
--- a/src/remote/attach.rs
+++ b/src/remote/attach.rs
@@ -86,7 +86,10 @@ pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
     run_client_process(&local_socket, &reattach_command, remote.keybindings)
 }
 
-pub(crate) fn prepare_saved_ssh(target: &str, session_name: &str) -> io::Result<()> {
+pub(crate) fn prepare_saved_ssh(
+    target: &str,
+    session_name: &str,
+) -> io::Result<Option<crate::client::endpoint::SshMachineMetadata>> {
     super::validate_remote_target(target)
         .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?;
     crate::session::validate_name(session_name)
@@ -135,7 +138,13 @@ pub(crate) fn prepare_saved_ssh(target: &str, session_name: &str) -> io::Result<
         )
         .is_none() =>
         {
-            Ok(())
+            Ok(prepared.remote_herdr.machine_metadata().or_else(|| {
+                discover_remote_api_metadata(&ssh, session_name)
+                    .inspect_err(
+                        |error| tracing::debug!(%error, "could not capture SSH setup metadata"),
+                    )
+                    .ok()
+            }))
         }
         _ => Err(io::Error::other(
             "remote server is not ready for saved machines",
@@ -318,11 +327,21 @@ impl RemoteExecutable {
 pub(super) struct RemoteHerdr {
     install_suffix: String,
     executable: RemoteExecutable,
+    resolved_executable: Option<String>,
     platform: RemotePlatform,
     bridge_idle_timeout: bool,
 }
 
 impl RemoteHerdr {
+    pub(super) fn machine_metadata(&self) -> Option<crate::client::endpoint::SshMachineMetadata> {
+        let executable = self.resolved_executable.clone()?;
+        let metadata = crate::client::endpoint::SshMachineMetadata {
+            os: self.platform.os.to_owned(),
+            executable,
+        };
+        metadata.is_valid().then_some(metadata)
+    }
+
     fn for_platform(platform: RemotePlatform) -> Self {
         let (install_suffix, executable) = if platform.is_windows() {
             (
@@ -337,17 +356,20 @@ impl RemoteHerdr {
         Self {
             install_suffix,
             executable,
+            resolved_executable: None,
             platform,
             bridge_idle_timeout: false,
         }
     }
 
-    fn with_shell_path(mut self, shell_path: String) -> Self {
-        self.executable = RemoteExecutable::PosixShellPath(shell_path);
+    fn with_posix_path(mut self, path: &str) -> Self {
+        self.executable = RemoteExecutable::PosixShellPath(shell_quote(path));
+        self.resolved_executable = Some(path.to_owned());
         self
     }
 
     fn with_windows_path(mut self, path: String) -> Self {
+        self.resolved_executable = Some(path.clone());
         self.executable = RemoteExecutable::WindowsPath(path);
         self
     }
@@ -364,6 +386,10 @@ fn windows_powershell_application_script(path: &str, args: &[&str]) -> String {
 }
 
 fn windows_powershell_streaming_application_command(path: &str, args: &[&str]) -> String {
+    windows_powershell_script_command(&windows_powershell_streaming_application_script(path, args))
+}
+
+fn windows_powershell_streaming_application_script(path: &str, args: &[&str]) -> String {
     let command_line = args
         .iter()
         .map(|arg| crate::platform::quote_windows_command_line_arg(arg))
@@ -371,11 +397,11 @@ fn windows_powershell_streaming_application_command(path: &str, args: &[&str]) -
         .join(" ");
     // Start-Process -Wait waits for descendants, including a cold-started server.
     // Retain the handle so Windows PowerShell 5.1 keeps the application's exit code.
-    windows_powershell_script_command(&format!(
+    format!(
         "$process = Start-Process -FilePath {} -ArgumentList {} -NoNewWindow -PassThru -ErrorAction Stop; $null = $process.Handle; $process.WaitForExit(); exit $process.ExitCode",
         crate::platform::quote_powershell_arg(path),
         crate::platform::quote_powershell_arg(&command_line),
-    ))
+    )
 }
 
 fn posix_remote_output_command(command: &str) -> String {
@@ -1248,11 +1274,29 @@ fn prepare_windows_remote_herdr(
     })
 }
 
-pub(super) fn find_installed_remote_api_herdr(
+pub(super) fn discover_remote_api_metadata(
     ssh: &RemoteSsh,
     session: &str,
-) -> io::Result<RemoteHerdr> {
+) -> io::Result<crate::client::endpoint::SshMachineMetadata> {
     let platform = detect_remote_platform(ssh)?;
+    if !platform.is_windows() {
+        let output =
+            ssh.framed_user_shell_output(&posix_remote_api_discovery_command(&platform, session))?;
+        if !output.status.success() {
+            return Err(command_failed("remote binary discovery failed", &output));
+        }
+        let metadata = crate::client::endpoint::SshMachineMetadata {
+            os: platform.os.to_owned(),
+            executable: String::from_utf8_lossy(&output.stdout).trim().to_owned(),
+        };
+        if !metadata.is_valid() {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "invalid remote Herdr executable path",
+            ));
+        }
+        return Ok(metadata);
+    }
     let remote_herdr = RemoteHerdr::for_platform(platform);
     let candidates = remote_binary_candidates(ssh, &remote_herdr)?;
     for candidate in candidates {
@@ -1264,7 +1308,10 @@ pub(super) fn find_installed_remote_api_herdr(
         if probe.status.success()
             && String::from_utf8_lossy(&probe.stdout).trim() == "herdr-api-bridge-v1"
         {
-            return Ok(candidate);
+            return Ok(crate::client::endpoint::SshMachineMetadata {
+                os: "windows".into(),
+                executable: candidate.executable.display().to_owned(),
+            });
         }
     }
     Err(io::Error::new(
@@ -1511,7 +1558,7 @@ fn remote_herdr_from_path(remote_herdr: &RemoteHerdr, path: &str) -> Option<Remo
     if is_mise_shim_path(path) {
         return None;
     }
-    Some(remote_herdr.clone().with_shell_path(shell_quote(path)))
+    Some(remote_herdr.clone().with_posix_path(path))
 }
 
 fn is_mise_shim_path(path: &str) -> bool {
@@ -2383,6 +2430,67 @@ fn confirm_remote_install(
     Ok(())
 }
 
+fn posix_remote_api_discovery_command(platform: &RemotePlatform, session: &str) -> String {
+    let script = format!(
+        r#"set -f
+candidates=$(
+command -v herdr
+{discovery}
+)
+IFS='
+'
+for candidate in $candidates; do
+    case "$candidate" in
+        */mise/shims/herdr) continue ;;
+        /*) ;;
+        *) continue ;;
+    esac
+    [ -x "$candidate" ] || continue
+    if capability=$("$candidate" --session {session} remote-api-bridge --check </dev/null 2>/dev/null) && [ "$capability" = herdr-api-bridge-v1 ]; then
+        printf '%s\n' "$candidate"
+        exit 0
+    fi
+done
+printf '%s\n' 'remote Herdr does not support machine API forwarding; update Herdr on this machine' >&2
+exit 2"#,
+        discovery = known_remote_binary_candidate_script(platform),
+        session = shell_quote(session),
+    );
+    format!(
+        "/bin/sh -c {}",
+        shell_quote(&posix_remote_output_command(&script))
+    )
+}
+
+pub(super) const STALE_API_METADATA: &str = "herdr-machine-metadata-stale-v1";
+
+pub(super) fn cached_remote_api_command(
+    metadata: &crate::client::endpoint::SshMachineMetadata,
+    session: &str,
+) -> String {
+    if metadata.os == "windows" {
+        let path = crate::platform::quote_powershell_arg(&metadata.executable);
+        let session_arg = crate::platform::quote_powershell_arg(session);
+        let probe = format!(
+            "$capability = & {path} --session {session_arg} remote-api-bridge --check 2>$null; if ($LASTEXITCODE -ne 0 -or $capability -ne 'herdr-api-bridge-v1') {{ [Console]::Error.WriteLine('{STALE_API_METADATA}'); exit 78 }}; "
+        );
+        return windows_powershell_script_command(&format!(
+            "{probe}{}",
+            windows_powershell_streaming_application_script(
+                &metadata.executable,
+                &["--session", session, "remote-api-bridge"]
+            ),
+        ));
+    }
+    let path = shell_quote(&metadata.executable);
+    let session = shell_quote(session);
+    let script = format!(
+        "if capability=$({path} --session {session} remote-api-bridge --check </dev/null 2>/dev/null) && [ \"$capability\" = herdr-api-bridge-v1 ]; then\n{}\nelse\n    printf '%s\\n' '{STALE_API_METADATA}' >&2\n    exit 78\nfi",
+        posix_remote_output_command(&format!("exec {path} --session {session} remote-api-bridge")),
+    );
+    format!("/bin/sh -c {}", shell_quote(&script))
+}
+
 pub(super) fn remote_api_bridge_command(
     remote_herdr: &RemoteHerdr,
     session_name: &str,
@@ -4070,6 +4178,63 @@ mod tests {
     }
 
     #[test]
+    fn machine_metadata_keeps_raw_resolved_paths_not_shell_expressions() {
+        let remote = RemoteHerdr::for_platform(RemotePlatform {
+            os: "linux",
+            arch: "x86_64",
+        });
+        assert!(remote.machine_metadata().is_none());
+        let path = "/home/user's files/$literal/herdr";
+        let resolved = remote.with_posix_path(path);
+        assert_eq!(resolved.machine_metadata().unwrap().executable, path);
+        assert_eq!(resolved.executable.display(), shell_quote(path));
+        let remote = RemoteHerdr::for_platform(RemotePlatform {
+            os: "windows",
+            arch: "x86_64",
+        });
+        assert!(remote.machine_metadata().is_none());
+        let path = r"C:\Users\A B\herdr.exe";
+        assert_eq!(
+            remote
+                .with_windows_path(path.into())
+                .machine_metadata()
+                .unwrap()
+                .executable,
+            path
+        );
+    }
+
+    #[test]
+    fn cached_windows_api_command_checks_before_starting_the_stream() {
+        let path = r"C:\Users\A'B\herdr.exe";
+        let command = cached_remote_api_command(
+            &crate::client::endpoint::SshMachineMetadata {
+                os: "windows".into(),
+                executable: path.into(),
+            },
+            "fleet",
+        );
+        let encoded = command.split_whitespace().last().unwrap();
+        let bytes = base64::engine::general_purpose::STANDARD
+            .decode(encoded)
+            .unwrap();
+        let words = bytes
+            .chunks_exact(2)
+            .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
+            .collect::<Vec<_>>();
+        let script = String::from_utf16(&words).unwrap();
+        assert!(script.contains(&crate::platform::quote_powershell_arg(path)));
+        assert!(script.contains(STALE_API_METADATA));
+        assert!(
+            script.find("remote-api-bridge --check").unwrap()
+                < script.find("Start-Process").unwrap()
+        );
+        assert!(script.contains("$LASTEXITCODE -ne 0"));
+        assert!(script.contains("-NoNewWindow -PassThru"));
+        assert!(script.contains("--session fleet remote-api-bridge"));
+    }
+
+    #[test]
     fn windows_remote_commands_use_one_encoded_powershell_grammar() {
         let executable = RemoteExecutable::WindowsPath("herdr.exe".to_string());
         let commands = [
diff --git a/src/remote/saved.rs b/src/remote/saved.rs
index 561ee1a0..4d8bd74b 100644
--- a/src/remote/saved.rs
+++ b/src/remote/saved.rs
@@ -19,6 +19,7 @@ pub(crate) fn connect_saved_ssh(
 ) -> io::Result<SavedSshStream> {
     let ssh = validated_saved_ssh(profile_id, target, session)?;
     let remote_herdr = find_installed_remote_herdr(&ssh)?;
+    let metadata = remote_herdr.machine_metadata();
     let path = saved_bridge_path(profile_id);
     let bridge = SshStdioBridge::start(
         target.to_owned(),
@@ -29,6 +30,10 @@ pub(crate) fn connect_saved_ssh(
         true,
     )?;
     let stream = crate::ipc::connect_local_stream(&path)?;
+    if let Some(metadata) = metadata {
+        crate::client::endpoint::SshMetadataCache::new(profile_id, target, session)?
+            .store(&metadata);
+    }
     Ok(SavedSshStream {
         stream,
         bridge: SavedSshBridge { _bridge: bridge },
@@ -38,13 +43,31 @@ pub(crate) fn connect_saved_ssh(
 pub(crate) struct SavedSshApiBridge {
     path: PathBuf,
     bridge: SshStdioBridge,
+    metadata_cache: crate::client::endpoint::SshMetadataCache,
+    pub(crate) used_cached_metadata: bool,
 }
 
 impl SavedSshApiBridge {
-    pub(crate) fn start(profile_id: &str, target: &str, session: &str) -> io::Result<Self> {
+    pub(crate) fn start(
+        profile_id: &str,
+        target: &str,
+        session: &str,
+        use_cached_metadata: bool,
+    ) -> io::Result<Self> {
         let ssh = validated_saved_ssh(profile_id, target, session)?;
-        let remote_herdr = super::attach::find_installed_remote_api_herdr(&ssh, session)?;
-        let command = super::attach::remote_api_bridge_command(&remote_herdr, session, false);
+        let metadata_cache =
+            crate::client::endpoint::SshMetadataCache::new(profile_id, target, session)?;
+        let cached = use_cached_metadata.then(|| metadata_cache.load()).flatten();
+        let used_cached_metadata = cached.is_some();
+        let metadata = match cached {
+            Some(metadata) => metadata,
+            None => {
+                let metadata = super::attach::discover_remote_api_metadata(&ssh, session)?;
+                metadata_cache.store(&metadata);
+                metadata
+            }
+        };
+        let command = super::attach::cached_remote_api_command(&metadata, session);
         let path = crate::platform::remote_bridge_endpoint_path(
             &format!("herdr-api-ssh-{}-{profile_id}.sock", std::process::id()),
             &format!(
@@ -60,7 +83,12 @@ impl SavedSshApiBridge {
             ssh.options(),
             true,
         )?;
-        Ok(Self { path, bridge })
+        Ok(Self {
+            path,
+            bridge,
+            metadata_cache,
+            used_cached_metadata,
+        })
     }
 
     pub(crate) fn socket_path(&self) -> &std::path::Path {
@@ -70,6 +98,16 @@ impl SavedSshApiBridge {
     pub(crate) fn reported_failure(&self) -> Option<io::Error> {
         self.bridge.reported_failure()
     }
+
+    pub(crate) fn invalidate_metadata(&self) {
+        self.metadata_cache.invalidate();
+    }
+
+    pub(crate) fn stale_metadata_failure(error: &io::Error) -> bool {
+        error
+            .to_string()
+            .contains(super::attach::STALE_API_METADATA)
+    }
 }
 
 pub(crate) fn saved_ssh_bootstrap_command(target: &str, session: &str) -> String {