fix: support preinstalled windows remote hosts (#3661)

* fix: support windows hosts in saved ssh machines

refs #3651

* fix: keep windows ssh bridge writes blocking

refs #3651

* fix: close saved windows ssh bootstrap input

refs #3651

* refactor: simplify remote bridge helpers

refs #3651

* refactor: isolate remote bridge platform behavior

refs #3651

* fix: close captured ssh input under windows consoles

refs #3651

* test: remove redundant remote command assertions

refs #3651

refs #3701

* test: handle framed ssh commands in machine setup

refs #3651

---------

Co-authored-by: Ogulcan Celik <ogulcancelik@gmail.com>

2 files changed, 658 insertions, 146 deletions

diff --git a/src/remote/attach.rs b/src/remote/attach.rs
index ac5415fa..2aee5efe 100644
--- a/src/remote/attach.rs
+++ b/src/remote/attach.rs
@@ -1,13 +1,16 @@
 //! Remote thin-client launcher over SSH command stdio.
 
 use super::{args::*, process::wait_with_output_timeout, restart_policy::*, shell_quote};
+use base64::Engine as _;
 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 _, Stream as _};
+use interprocess::local_socket::traits::Listener as _;
+#[cfg(all(test, unix))]
+use interprocess::local_socket::traits::Stream as _;
 use interprocess::local_socket::ListenerNonblockingMode;
 use interprocess::TryClone as _;
 use serde::Deserialize;
@@ -30,6 +33,7 @@ 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 REMOTE_OUTPUT_READY_MARKER: &str = "herdr-remote-output-ready:1";
 const SSH_CONTROL_SOCKET_NAME: &str = "ctl";
 pub(crate) fn run_remote(remote: RemoteLaunch) -> io::Result<()> {
     let session_name = crate::session::active_name()
@@ -104,10 +108,11 @@ pub(crate) fn prepare_saved_ssh(target: &str, session_name: &str) -> io::Result<
 
     // 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)
-    ))?;
+    let command = prepared
+        .remote_herdr
+        .executable
+        .saved_bridge_command(session_name);
+    let output = ssh.shell_output(&prepared.remote_herdr.platform, &command)?;
     if !output.status.success() {
         return Err(command_failed("remote server startup failed", &output));
     }
@@ -161,6 +166,8 @@ impl RemotePlatform {
             "linux"
         } else if cfg!(target_os = "macos") {
             "macos"
+        } else if cfg!(target_os = "windows") {
+            "windows"
         } else {
             "unknown"
         };
@@ -179,32 +186,193 @@ impl RemotePlatform {
     fn asset_key(&self) -> String {
         format!("{}-{}", self.os, self.arch)
     }
+
+    fn is_windows(&self) -> bool {
+        self.os == "windows"
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum RemoteExecutable {
+    PosixShellPath(String),
+    WindowsPath(String),
+}
+
+impl RemoteExecutable {
+    fn display(&self) -> &str {
+        match self {
+            Self::PosixShellPath(path) | Self::WindowsPath(path) => path,
+        }
+    }
+
+    fn command(&self, args: &[&str]) -> String {
+        match self {
+            Self::PosixShellPath(path) => {
+                let mut command = path.clone();
+                for arg in args {
+                    command.push(' ');
+                    command.push_str(&shell_quote(arg));
+                }
+                command
+            }
+            Self::WindowsPath(path) => windows_powershell_script_command(
+                &windows_powershell_application_script(path, args),
+            ),
+        }
+    }
+
+    fn session_command(&self, session_name: &str, args: &[&str]) -> String {
+        self.command(&Self::session_args(session_name, args))
+    }
+
+    fn session_args<'a>(session_name: &'a str, args: &[&'a str]) -> Vec<&'a str> {
+        let mut session_args = Vec::with_capacity(args.len() + 2);
+        if session_name != crate::session::DEFAULT_SESSION_NAME {
+            session_args.extend(["--session", session_name]);
+        }
+        session_args.extend_from_slice(args);
+        session_args
+    }
+
+    fn exists_command(&self) -> String {
+        match self {
+            Self::PosixShellPath(path) => format!("test -x {path}"),
+            Self::WindowsPath(path) => windows_powershell_script_command(&format!(
+                "if ($null -ne (Get-Command {} -CommandType Application -ErrorAction SilentlyContinue)) {{ exit 0 }}; exit 1",
+                crate::platform::quote_powershell_arg(path)
+            )),
+        }
+    }
+
+    fn status_client_command(&self) -> String {
+        match self {
+            Self::PosixShellPath(path) => format!(
+                "test -x {path} && {}",
+                self.command(&["status", "client", "--json"])
+            ),
+            Self::WindowsPath(_) => self.command(&["status", "client", "--json"]),
+        }
+    }
+
+    fn bridge_command(&self, session_name: &str) -> String {
+        let args = Self::session_args(session_name, &["remote-client-bridge"]);
+        match self {
+            Self::PosixShellPath(_) => {
+                posix_remote_output_command(&format!("exec {}", self.command(&args)))
+            }
+            Self::WindowsPath(path) => {
+                windows_powershell_streaming_application_command(path, &args)
+            }
+        }
+    }
+
+    fn saved_bridge_command(&self, session_name: &str) -> String {
+        let args = Self::session_args(session_name, &["remote-client-bridge"]);
+        match self {
+            Self::PosixShellPath(_) => format!("exec {} </dev/null", self.command(&args)),
+            Self::WindowsPath(path) => {
+                windows_powershell_streaming_application_command(path, &args)
+            }
+        }
+    }
+
+    fn live_handoff_command(&self, session_name: &str, protocol: u32, version: &str) -> String {
+        let protocol = protocol.to_string();
+        match self {
+            Self::PosixShellPath(path) => format!(
+                "{} --import-exe {path} --expected-protocol {} --expected-version {}",
+                self.session_command(session_name, &["server", "live-handoff"]),
+                shell_quote(&protocol),
+                shell_quote(version),
+            ),
+            Self::WindowsPath(path) => self.session_command(
+                session_name,
+                &[
+                    "server",
+                    "live-handoff",
+                    "--import-exe",
+                    path,
+                    "--expected-protocol",
+                    &protocol,
+                    "--expected-version",
+                    version,
+                ],
+            ),
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
 pub(super) struct RemoteHerdr {
     install_suffix: String,
-    shell_path: String,
+    executable: RemoteExecutable,
     platform: RemotePlatform,
 }
 
 impl RemoteHerdr {
     fn for_platform(platform: RemotePlatform) -> Self {
-        let install_suffix = ".local/bin/herdr".to_string();
-        let shell_path = format!("\"$HOME/{install_suffix}\"");
+        let (install_suffix, executable) = if platform.is_windows() {
+            (
+                String::new(),
+                RemoteExecutable::WindowsPath("herdr.exe".to_string()),
+            )
+        } else {
+            let install_suffix = ".local/bin/herdr".to_string();
+            let shell_path = format!("\"$HOME/{install_suffix}\"");
+            (install_suffix, RemoteExecutable::PosixShellPath(shell_path))
+        };
         Self {
             install_suffix,
-            shell_path,
+            executable,
             platform,
         }
     }
 
     fn with_shell_path(mut self, shell_path: String) -> Self {
-        self.shell_path = shell_path;
+        self.executable = RemoteExecutable::PosixShellPath(shell_path);
         self
     }
 }
 
+fn windows_powershell_application_script(path: &str, args: &[&str]) -> String {
+    let mut script = format!("& {}", crate::platform::quote_powershell_arg(path));
+    for arg in args {
+        script.push(' ');
+        script.push_str(&crate::platform::quote_powershell_arg(arg));
+    }
+    script.push_str("; exit $LASTEXITCODE");
+    script
+}
+
+fn windows_powershell_streaming_application_command(path: &str, args: &[&str]) -> String {
+    let command_line = args
+        .iter()
+        .map(|arg| crate::platform::quote_windows_command_line_arg(arg))
+        .collect::<Vec<_>>()
+        .join(" ");
+    windows_powershell_script_command(&format!(
+        "$process = Start-Process -FilePath {} -ArgumentList {} -NoNewWindow -Wait -PassThru -ErrorAction Stop; exit $process.ExitCode",
+        crate::platform::quote_powershell_arg(path),
+        crate::platform::quote_powershell_arg(&command_line),
+    ))
+}
+
+fn posix_remote_output_command(command: &str) -> String {
+    format!("printf '\n%s\n' '{REMOTE_OUTPUT_READY_MARKER}'\n{command}")
+}
+
+fn windows_powershell_script_command(script: &str) -> String {
+    let script = format!(
+        "[Console]::Out.WriteLine(); [Console]::Out.WriteLine('{REMOTE_OUTPUT_READY_MARKER}'); [Console]::Out.Flush(); {script}"
+    );
+    let utf16 = script
+        .encode_utf16()
+        .flat_map(u16::to_le_bytes)
+        .collect::<Vec<_>>();
+    let encoded = base64::engine::general_purpose::STANDARD.encode(utf16);
+    format!("powershell.exe -NoLogo -NoProfile -NonInteractive -EncodedCommand {encoded}")
+}
+
 #[derive(Debug, Clone, Deserialize)]
 #[serde(untagged)]
 enum RemoteAssetRef {
@@ -417,6 +585,7 @@ impl RemoteSsh {
     }
 
     fn sh_output(&self, script: &str) -> io::Result<Output> {
+        let script = posix_remote_output_command(script);
         let mut child = self
             .command()
             .arg("/bin/sh -s")
@@ -439,19 +608,35 @@ impl RemoteSsh {
             child.wait_with_output()?
         };
         write_result?;
-        Ok(output)
+        normalize_remote_output(output)
     }
 
-    fn user_shell_output(&self, remote_command: &str) -> io::Result<Output> {
+    fn framed_user_shell_output(&self, remote_command: &str) -> io::Result<Output> {
         let mut command = self.command();
         command
+            // Windows OpenSSH can still read the console with stdin redirected to NUL.
+            .arg("-n")
             .arg(remote_command)
+            .stdin(Stdio::null())
             .stdout(Stdio::piped())
             .stderr(Stdio::piped());
-        if self.noninteractive {
+        let output = if self.noninteractive {
             wait_with_output_timeout(command.spawn()?, NONINTERACTIVE_SSH_COMMAND_TIMEOUT)
         } else {
             command.output()
+        }?;
+        normalize_remote_output(output)
+    }
+
+    fn posix_user_shell_output(&self, remote_command: &str) -> io::Result<Output> {
+        self.framed_user_shell_output(&posix_remote_output_command(remote_command))
+    }
+
+    fn shell_output(&self, platform: &RemotePlatform, remote_command: &str) -> io::Result<Output> {
+        if platform.is_windows() {
+            self.framed_user_shell_output(remote_command)
+        } else {
+            self.sh_output(remote_command)
         }
     }
 
@@ -500,6 +685,24 @@ impl RemoteSsh {
     }
 }
 
+fn normalize_remote_output(mut output: Output) -> io::Result<Output> {
+    normalize_remote_stdout(&mut output.stdout, output.status.success())?;
+    Ok(output)
+}
+
+fn normalize_remote_stdout(stdout: &mut Vec<u8>, command_succeeded: bool) -> io::Result<()> {
+    let consumed = {
+        let mut reader = io::Cursor::new(stdout.as_slice());
+        match discard_remote_output_preamble(&mut reader) {
+            Ok(()) => reader.position() as usize,
+            Err(_) if !command_succeeded => return Ok(()),
+            Err(err) => return Err(err),
+        }
+    };
+    stdout.drain(..consumed);
+    Ok(())
+}
+
 fn remote_install_prepare_script(remote_herdr: &RemoteHerdr) -> String {
     format!(
         r#"set -eu
@@ -636,6 +839,9 @@ pub(super) fn prepare_remote_herdr(
 ) -> io::Result<PreparedRemoteHerdr> {
     let platform = detect_remote_platform(ssh)?;
     let remote_herdr = RemoteHerdr::for_platform(platform);
+    if remote_herdr.platform.is_windows() {
+        return prepare_windows_remote_herdr(ssh, remote_herdr, require_surface_interest);
+    }
     let override_binary = remote_binary_override_path()?;
     let remote_binary_candidates = remote_binary_candidates(ssh, &remote_herdr)?;
 
@@ -690,7 +896,7 @@ pub(super) fn prepare_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 saved SSH endpoint federation",
-            remote_herdr.shell_path
+            remote_herdr.executable.display()
         )));
     }
     warn_if_remote_bin_not_on_path(ssh)?;
@@ -704,6 +910,10 @@ pub(super) 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);
+    if remote_herdr.platform.is_windows() {
+        return prepare_windows_remote_herdr(ssh, remote_herdr, true)
+            .map(|prepared| prepared.remote_herdr);
+    }
     let candidates = remote_binary_candidates(ssh, &remote_herdr)?;
     for candidate in candidates {
         if remote_binary_supports_endpoint_requirement(ssh, &candidate, true)? {
@@ -720,23 +930,107 @@ pub(super) fn find_installed_remote_herdr(ssh: &RemoteSsh) -> io::Result<RemoteH
     ))
 }
 
-fn detect_remote_platform(ssh: &RemoteSsh) -> io::Result<RemotePlatform> {
-    let output = ssh.sh_output("uname -s\nuname -m\n")?;
-    if !output.status.success() {
-        return Err(command_failed("remote platform detection failed", &output));
+fn prepare_windows_remote_herdr(
+    ssh: &RemoteSsh,
+    remote_herdr: RemoteHerdr,
+    require_surface_interest: bool,
+) -> io::Result<PreparedRemoteHerdr> {
+    if !remote_binary_exists(ssh, &remote_herdr)? {
+        return Err(io::Error::new(
+            io::ErrorKind::NotFound,
+            format!(
+                "herdr.exe is not installed or is not on PATH on Windows host {}; install a compatible Windows package with remote host support and retry",
+                ssh.target()
+            ),
+        ));
+    }
+    if !remote_binary_supports_endpoint_requirement(ssh, &remote_herdr, require_surface_interest)? {
+        return Err(io::Error::other(format!(
+            "herdr.exe on Windows host {} does not support saved SSH endpoint federation; install a compatible Windows package with remote host support and retry",
+            ssh.target()
+        )));
     }
 
-    let stdout = String::from_utf8_lossy(&output.stdout);
-    let mut lines = stdout.lines();
-    let os = lines.next().unwrap_or_default();
-    let arch = lines.next().unwrap_or_default();
-    RemotePlatform::from_uname(os, arch).ok_or_else(|| {
+    Ok(PreparedRemoteHerdr {
+        remote_herdr,
+        stop_after_install_approved: false,
+    })
+}
+
+fn detect_remote_platform(ssh: &RemoteSsh) -> io::Result<RemotePlatform> {
+    let output = ssh.sh_output("uname -s\nuname -m\n")?;
+    let mut windows_uname_hint = false;
+    let posix_error = if output.status.success() {
+        let stdout = String::from_utf8_lossy(&output.stdout);
+        let mut lines = stdout.lines();
+        let os = lines.next().unwrap_or_default();
+        let arch = lines.next().unwrap_or_default();
+        if let Some(platform) = RemotePlatform::from_uname(os, arch) {
+            return Ok(platform);
+        }
+        windows_uname_hint = looks_like_windows_uname(os);
         io::Error::other(format!(
             "unsupported remote platform: {} {}",
             os.trim(),
             arch.trim()
         ))
-    })
+    } else {
+        command_failed("remote platform detection failed", &output)
+    };
+
+    if !should_try_windows_probe(
+        output.status.success(),
+        output.status.code(),
+        windows_uname_hint,
+    ) {
+        return Err(posix_error);
+    }
+    let windows_output = match ssh.framed_user_shell_output(&windows_platform_probe_command()) {
+        Ok(output) if output.status.success() => output,
+        _ => return Err(posix_error),
+    };
+    match parse_windows_platform_probe(&String::from_utf8_lossy(&windows_output.stdout)) {
+        Ok(Some(platform)) => Ok(platform),
+        Ok(None) => Err(posix_error),
+        Err(message) => Err(io::Error::other(message)),
+    }
+}
+
+fn should_try_windows_probe(
+    success: bool,
+    exit_code: Option<i32>,
+    windows_uname_hint: bool,
+) -> bool {
+    (success && windows_uname_hint) || (!success && exit_code.is_some_and(|code| code != 255))
+}
+
+fn looks_like_windows_uname(os: &str) -> bool {
+    let os = os.trim().to_ascii_uppercase();
+    ["MINGW", "MSYS", "CYGWIN"]
+        .iter()
+        .any(|prefix| os.starts_with(prefix))
+}
+
+fn windows_platform_probe_command() -> String {
+    windows_powershell_script_command(
+        "$arch = if ($env:PROCESSOR_ARCHITEW6432) { $env:PROCESSOR_ARCHITEW6432 } else { $env:PROCESSOR_ARCHITECTURE }; [Console]::Out.WriteLine('herdr-windows:' + $arch); exit 0",
+    )
+}
+
+fn parse_windows_platform_probe(stdout: &str) -> Result<Option<RemotePlatform>, String> {
+    let Some(arch) = stdout
+        .lines()
+        .find_map(|line| line.trim().strip_prefix("herdr-windows:"))
+    else {
+        return Ok(None);
+    };
+    match arch.trim().to_ascii_uppercase().as_str() {
+        "AMD64" | "X86_64" => Ok(Some(RemotePlatform {
+            os: "windows",
+            arch: "x86_64",
+        })),
+        arch => Err(format!("unsupported remote platform: Windows {arch}")),
+    }
 }
 
 fn remote_binary_candidates(
@@ -766,7 +1060,7 @@ fn remote_binary_candidates(
 fn push_if_new_remote_binary_candidate(candidates: &mut Vec<RemoteHerdr>, candidate: RemoteHerdr) {
     if !candidates
         .iter()
-        .any(|existing| existing.shell_path == candidate.shell_path)
+        .any(|existing| existing.executable == candidate.executable)
     {
         candidates.push(candidate);
     }
@@ -826,7 +1120,7 @@ fn remote_binary_on_path_any(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
 ) -> io::Result<Option<RemoteHerdr>> {
-    let output = ssh.user_shell_output("command -v herdr")?;
+    let output = ssh.posix_user_shell_output("command -v herdr")?;
     if output.status.success() {
         let stdout = String::from_utf8_lossy(&output.stdout);
         if let Some(candidate) = remote_herdr_from_path_discovery(remote_herdr, &stdout) {
@@ -880,11 +1174,8 @@ fn remote_client_status(
     ssh: &RemoteSsh,
     remote_herdr: &RemoteHerdr,
 ) -> io::Result<Option<RemoteClientStatusJson>> {
-    let command = format!(
-        "test -x {0} && {0} status client --json",
-        remote_herdr.shell_path
-    );
-    let output = ssh.sh_output(&command)?;
+    let command = remote_herdr.executable.status_client_command();
+    let output = ssh.shell_output(&remote_herdr.platform, &command)?;
     if !output.status.success() {
         if output.status.code() == Some(255) {
             return Err(command_failed("remote SSH connection failed", &output));
@@ -906,8 +1197,11 @@ fn remote_binary_supports_endpoint_requirement(
 }
 
 fn remote_binary_exists(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<bool> {
-    let command = format!("test -x {}", remote_herdr.shell_path);
-    Ok(ssh.sh_output(&command)?.status.success())
+    let command = remote_herdr.executable.exists_command();
+    Ok(ssh
+        .shell_output(&remote_herdr.platform, &command)?
+        .status
+        .success())
 }
 
 fn remote_binary_override_path() -> io::Result<Option<PathBuf>> {
@@ -1203,8 +1497,10 @@ fn remote_server_status(
     remote_herdr: &RemoteHerdr,
     require_surface_interest: bool,
 ) -> io::Result<RemoteServerStatus> {
-    let command = remote_session_command(remote_herdr, &ssh.session_name, "status server --json");
-    let output = ssh.sh_output(&command)?;
+    let command = remote_herdr
+        .executable
+        .session_command(&ssh.session_name, &["status", "server", "--json"]);
+    let output = ssh.shell_output(&remote_herdr.platform, &command)?;
     if !output.status.success() {
         return Err(command_failed("remote server status failed", &output));
     }
@@ -1425,22 +1721,6 @@ fn confirm_remote_server_stop(
     Ok(false)
 }
 
-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
-        ),
-    )
-}
-
 fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
     let status = remote_client_status(ssh, remote_herdr)?.ok_or_else(|| {
         io::Error::other("could not inspect the prepared remote herdr binary before live handoff")
@@ -1452,8 +1732,11 @@ 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, &ssh.session_name, protocol, &version);
-    let output = ssh.sh_output(&command)?;
+    let command =
+        remote_herdr
+            .executable
+            .live_handoff_command(&ssh.session_name, protocol, &version);
+    let output = ssh.shell_output(&remote_herdr.platform, &command)?;
     if !output.status.success() {
         return Err(command_failed("remote server live handoff failed", &output));
     }
@@ -1466,8 +1749,10 @@ fn live_handoff_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io
 }
 
 fn stop_remote_server(ssh: &RemoteSsh, remote_herdr: &RemoteHerdr) -> io::Result<()> {
-    let command = remote_session_command(remote_herdr, &ssh.session_name, "server stop");
-    let output = ssh.sh_output(&command)?;
+    let command = remote_herdr
+        .executable
+        .session_command(&ssh.session_name, &["server", "stop"]);
+    let output = ssh.shell_output(&remote_herdr.platform, &command)?;
     if !output.status.success() {
         return Err(command_failed("remote server stop failed", &output));
     }
@@ -1505,7 +1790,7 @@ fn version_label(version: Option<&str>) -> &str {
 }
 
 fn warn_if_remote_bin_not_on_path(ssh: &RemoteSsh) -> io::Result<()> {
-    let output = ssh.user_shell_output("command -v herdr")?;
+    let output = ssh.posix_user_shell_output("command -v herdr")?;
     if output.status.success()
         && remote_shell_resolves_managed_install(&String::from_utf8_lossy(&output.stdout))
     {
@@ -1704,7 +1989,7 @@ fn confirm_remote_install(
         return Err(io::Error::other(format!(
             "matching remote herdr {} is not installed at {}; run from an interactive terminal to approve installation",
             current_version(),
-            remote_herdr.shell_path
+            remote_herdr.executable.display()
         )));
     }
 
@@ -1715,7 +2000,8 @@ fn confirm_remote_install(
     );
     eprint!(
         "Install {} to {}? [Y/n] ",
-        source_description, remote_herdr.shell_path
+        source_description,
+        remote_herdr.executable.display()
     );
     io::stderr().flush()?;
 
@@ -1729,24 +2015,6 @@ fn confirm_remote_install(
     Ok(())
 }
 
-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(' ');
-    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,
@@ -2015,7 +2283,7 @@ pub(crate) fn bridge_upload_cancellation_for_test(
 }
 
 fn bridge_connection(
-    stream: crate::ipc::LocalStream,
+    mut stream: crate::ipc::LocalStream,
     target: &str,
     remote_herdr: &RemoteHerdr,
     session_name: &str,
@@ -2032,7 +2300,7 @@ fn bridge_connection(
     command
         .arg("-T")
         .arg(target)
-        .arg(remote_bridge_command(remote_herdr, session_name))
+        .arg(remote_herdr.executable.bridge_command(session_name))
         .stdin(Stdio::piped())
         .stdout(Stdio::piped())
         .stderr(if noninteractive {
@@ -2048,7 +2316,7 @@ fn bridge_connection(
         Some(stdin) => stdin,
         None => return terminate_bridge_child(child, "ssh bridge stdin missing"),
     };
-    let mut child_stdout = match child.stdout.take() {
+    let child_stdout = match child.stdout.take() {
         Some(stdout) => stdout,
         None => return terminate_bridge_child(child, "ssh bridge stdout missing"),
     };
@@ -2068,7 +2336,7 @@ fn bridge_connection(
             return Err(err);
         }
     };
-    if let Err(err) = stream.set_nonblocking(true) {
+    if let Err(err) = crate::ipc::set_local_stream_polling(&mut stream, true) {
         let _ = child.kill();
         let _ = child.wait();
         return Err(err);
@@ -2099,12 +2367,15 @@ fn bridge_connection(
     let download_done_worker = Arc::clone(&download_done);
     let download_upload_stop = Arc::clone(&upload_stop);
     let download = thread::spawn(move || {
-        let result = copy_reader_to_local_stream(
-            &mut child_stdout,
-            &mut child_to_stream,
-            &download_stop,
-            &download_bridge_stop,
-        );
+        let mut child_stdout = io::BufReader::new(child_stdout);
+        let result = discard_remote_output_preamble(&mut child_stdout).and_then(|()| {
+            copy_reader_to_local_stream(
+                &mut child_stdout,
+                &mut child_to_stream,
+                &download_stop,
+                &download_bridge_stop,
+            )
+        });
         download_done_worker.store(true, Ordering::Release);
         download_upload_stop.cancel();
         result
@@ -2209,6 +2480,49 @@ fn capture_ssh_stderr(mut stderr: impl io::Read) -> io::Result<Vec<u8>> {
     }
 }
 
+fn discard_remote_output_preamble(reader: &mut impl io::BufRead) -> io::Result<()> {
+    let marker = REMOTE_OUTPUT_READY_MARKER.as_bytes();
+    let mut matched = 0;
+    let mut matching = true;
+    loop {
+        let (consumed, ready) = {
+            let buffer = reader.fill_buf()?;
+            if buffer.is_empty() {
+                if matching && matched == marker.len() {
+                    return Ok(());
+                }
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    "remote command exited before producing its output marker",
+                ));
+            }
+
+            let mut consumed = 0;
+            let mut ready = false;
+            for &byte in buffer {
+                consumed += 1;
+                if byte == b'\n' {
+                    if matching && matched == marker.len() {
+                        ready = true;
+                        break;
+                    }
+                    matched = 0;
+                    matching = true;
+                } else if matching && matched < marker.len() && byte == marker[matched] {
+                    matched += 1;
+                } else if matching && (matched != marker.len() || byte != b'\r') {
+                    matching = false;
+                }
+            }
+            (consumed, ready)
+        };
+        reader.consume(consumed);
+        if ready {
+            return Ok(());
+        }
+    }
+}
+
 fn terminate_bridge_child(mut child: std::process::Child, message: &'static str) -> io::Result<()> {
     let _ = child.kill();
     let _ = child.wait();
@@ -2578,6 +2892,67 @@ mod tests {
         let _ = std::fs::remove_file(socket);
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn bridge_stream_delivers_large_frame_before_delayed_reply() {
+        use std::io::Read as _;
+        use std::sync::mpsc;
+        use std::time::{SystemTime, UNIX_EPOCH};
+
+        fn read_frame(stream: &mut crate::ipc::LocalStream) -> Vec<u8> {
+            let mut length = [0; 4];
+            stream.read_exact(&mut length).expect("read frame length");
+            let length = usize::try_from(u32::from_le_bytes(length)).expect("frame length fits");
+            let mut body = vec![0; length];
+            stream.read_exact(&mut body).expect("read frame body");
+            body
+        }
+
+        fn write_frame(stream: &mut crate::ipc::LocalStream, body: &[u8]) {
+            let length = u32::try_from(body.len()).expect("frame length fits");
+            stream
+                .write_all(&length.to_le_bytes())
+                .expect("write frame length");
+            stream.write_all(body).expect("write frame body");
+        }
+
+        let nonce = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .expect("system clock after epoch")
+            .as_nanos();
+        let socket = std::env::temp_dir().join(format!(
+            "herdr-bridge-large-frame-{}-{nonce}.sock",
+            std::process::id()
+        ));
+        let listener = crate::ipc::bind_private_local_listener(&socket).expect("bind listener");
+        let (large_received_tx, large_received_rx) = mpsc::channel();
+        let client_socket = socket.clone();
+        let client = thread::spawn(move || {
+            let mut stream =
+                crate::ipc::connect_local_stream(&client_socket).expect("connect client");
+            assert_eq!(read_frame(&mut stream), b"welcome");
+            assert_eq!(read_frame(&mut stream), vec![b'x'; 16 * 1024]);
+            large_received_tx.send(()).expect("signal large frame");
+            assert_eq!(read_frame(&mut stream), b"pong");
+        });
+        let mut server = prepare_remote_bridge_stream(listener.accept().expect("accept client"))
+            .expect("prepare bridge stream");
+
+        crate::ipc::set_local_stream_polling(&mut server, true)
+            .expect("enable bridge read polling");
+        write_frame(&mut server, b"welcome");
+        write_frame(&mut server, &vec![b'x'; 16 * 1024]);
+        large_received_rx
+            .recv_timeout(Duration::from_secs(2))
+            .expect("client received large frame");
+        write_frame(&mut server, b"pong");
+
+        client.join().expect("client thread");
+        drop(server);
+        drop(listener);
+        let _ = std::fs::remove_file(socket);
+    }
+
     #[test]
     fn bridge_drop_while_waiting_for_client_is_bounded() {
         let socket = local_forward_socket_path("drop-test", "default");
@@ -2835,26 +3210,29 @@ mod tests {
     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",
+        for (args, command) in [
+            (&["status", "server", "--json"][..], "status server --json"),
+            (&["server", "stop"][..], "server stop"),
+            (&["remote-client-bridge"][..], "remote-client-bridge"),
         ] {
             assert_eq!(
-                remote_session_command(&herdr, "agents", command),
-                format!("{} --session agents {command}", herdr.shell_path)
+                herdr.executable.session_command("agents", args),
+                format!("{} --session agents {command}", herdr.executable.display())
             );
             assert_eq!(
-                remote_session_command(&herdr, crate::session::DEFAULT_SESSION_NAME, command),
-                format!("{} {command}", herdr.shell_path)
+                herdr
+                    .executable
+                    .session_command(crate::session::DEFAULT_SESSION_NAME, args),
+                format!("{} {command}", herdr.executable.display())
             );
         }
-        assert!(
-            remote_live_handoff_command(&herdr, "agents", 19, "0.7.9").starts_with(&format!(
+        assert!(herdr
+            .executable
+            .live_handoff_command("agents", 19, "0.7.9")
+            .starts_with(&format!(
                 "{} --session agents server live-handoff",
-                herdr.shell_path
-            ))
-        );
+                herdr.executable.display()
+            )));
     }
 
     #[test]
@@ -3081,6 +3459,147 @@ mod tests {
         assert!(RemotePlatform::from_uname("FreeBSD", "x86_64").is_none());
     }
 
+    #[test]
+    fn windows_platform_probe_accepts_only_x86_64() {
+        assert_eq!(
+            parse_windows_platform_probe("profile noise\r\nherdr-windows:AMD64\r\n").unwrap(),
+            Some(RemotePlatform {
+                os: "windows",
+                arch: "x86_64",
+            })
+        );
+        assert_eq!(parse_windows_platform_probe("other output").unwrap(), None);
+        assert_eq!(
+            parse_windows_platform_probe("herdr-windows:ARM64").unwrap_err(),
+            "unsupported remote platform: Windows ARM64"
+        );
+    }
+
+    #[test]
+    fn windows_probe_preserves_transport_and_unsupported_unix_failures() {
+        let cases = [
+            (true, Some(0), false, false),
+            (true, Some(0), true, true),
+            (false, Some(255), false, false),
+            (false, None, false, false),
+            (false, Some(1), false, true),
+        ];
+        for (success, exit_code, windows_uname_hint, expected) in cases {
+            assert_eq!(
+                should_try_windows_probe(success, exit_code, windows_uname_hint),
+                expected,
+                "success={success}, exit_code={exit_code:?}, windows_uname_hint={windows_uname_hint}"
+            );
+        }
+
+        for (os, expected) in [
+            ("MINGW64_NT-10.0-26100", true),
+            ("MSYS_NT-10.0", true),
+            ("CYGWIN_NT-10.0", true),
+            ("FreeBSD", false),
+        ] {
+            assert_eq!(looks_like_windows_uname(os), expected, "{os}");
+        }
+    }
+
+    #[test]
+    fn windows_remote_commands_use_one_encoded_powershell_grammar() {
+        fn decode(command: &str) -> String {
+            let encoded = command
+                .strip_prefix("powershell.exe -NoLogo -NoProfile -NonInteractive -EncodedCommand ")
+                .expect("encoded PowerShell command");
+            let bytes = base64::engine::general_purpose::STANDARD
+                .decode(encoded)
+                .expect("base64");
+            let utf16 = bytes
+                .chunks_exact(2)
+                .map(|pair| u16::from_le_bytes([pair[0], pair[1]]))
+                .collect::<Vec<_>>();
+            String::from_utf16(&utf16).expect("UTF-16LE")
+        }
+
+        let executable = RemoteExecutable::WindowsPath("herdr.exe".to_string());
+        let commands = [
+            (
+                "platform probe",
+                windows_platform_probe_command(),
+                "$arch = if ($env:PROCESSOR_ARCHITEW6432) { $env:PROCESSOR_ARCHITEW6432 } else { $env:PROCESSOR_ARCHITECTURE }; [Console]::Out.WriteLine('herdr-windows:' + $arch); exit 0",
+            ),
+            (
+                "PATH lookup",
+                executable.exists_command(),
+                "if ($null -ne (Get-Command herdr.exe -CommandType Application -ErrorAction SilentlyContinue)) { exit 0 }; exit 1",
+            ),
+            (
+                "client status",
+                executable.status_client_command(),
+                "& herdr.exe status client '--json'; exit $LASTEXITCODE",
+            ),
+            (
+                "named server status",
+                executable.session_command("agents", &["status", "server", "--json"]),
+                "& herdr.exe '--session' agents status server '--json'; exit $LASTEXITCODE",
+            ),
+            (
+                "server stop",
+                executable.session_command("agents", &["server", "stop"]),
+                "& herdr.exe '--session' agents server stop; exit $LASTEXITCODE",
+            ),
+            (
+                "direct bridge",
+                executable.bridge_command("agents"),
+                "$process = Start-Process -FilePath herdr.exe -ArgumentList '--session agents remote-client-bridge' -NoNewWindow -Wait -PassThru -ErrorAction Stop; exit $process.ExitCode",
+            ),
+            (
+                "saved bridge with closed stdin",
+                executable.saved_bridge_command("agents"),
+                "$process = Start-Process -FilePath herdr.exe -ArgumentList '--session agents remote-client-bridge' -NoNewWindow -Wait -PassThru -ErrorAction Stop; exit $process.ExitCode",
+            ),
+        ];
+
+        let preamble = format!(
+            "[Console]::Out.WriteLine(); [Console]::Out.WriteLine('{REMOTE_OUTPUT_READY_MARKER}'); [Console]::Out.Flush(); "
+        );
+        for (label, command, expected) in commands {
+            let script = decode(&command);
+            let script = script
+                .strip_prefix(&preamble)
+                .unwrap_or_else(|| panic!("{label} lacks shared output marker: {script}"));
+            assert_eq!(script, expected, "{label}");
+        }
+    }
+
+    #[test]
+    fn remote_output_framing_discards_any_banner_and_preserves_binary() {
+        let payload = [0, 1, 2, 0xff, b'\n'];
+        let mut input = vec![b'x'; 4 * 1024 * 1024];
+        input.extend_from_slice(b"\r\nherdr-remote-output-ready:1\r\n");
+        input.extend_from_slice(&payload);
+        let mut reader = io::BufReader::with_capacity(17, io::Cursor::new(input));
+
+        discard_remote_output_preamble(&mut reader).unwrap();
+        let mut output = Vec::new();
+        io::Read::read_to_end(&mut reader, &mut output).unwrap();
+        assert_eq!(output, payload);
+
+        let mut missing = b"profile output without marker".to_vec();
+        assert!(normalize_remote_stdout(&mut missing, true).is_err());
+        normalize_remote_stdout(&mut missing, false).unwrap();
+        assert_eq!(missing, b"profile output without marker");
+
+        let mut platform = b"profile output\nherdr-remote-output-ready:1\nLinux\nx86_64\n".to_vec();
+        normalize_remote_stdout(&mut platform, true).unwrap();
+        let platform = String::from_utf8(platform).unwrap();
+        let mut lines = platform.lines();
+        assert_eq!(
+            RemotePlatform::from_uname(lines.next().unwrap(), lines.next().unwrap()),
+            Some(RemotePlatform {
+                os: "linux",
+                arch: "x86_64"
+            })
+        );
+    }
+
     #[cfg(unix)]
     #[test]
     fn reattach_command_includes_remote_and_session() {
@@ -3152,8 +3671,14 @@ mod tests {
             arch: "x86_64",
         });
         assert_eq!(
-            remote_bridge_command(&remote_herdr, crate::session::DEFAULT_SESSION_NAME),
-            "exec \"$HOME/.local/bin/herdr\" remote-client-bridge"
+            remote_herdr
+                .executable
+                .bridge_command(crate::session::DEFAULT_SESSION_NAME),
+            "printf '\n%s\n' 'herdr-remote-output-ready:1'\nexec \"$HOME/.local/bin/herdr\" remote-client-bridge"
+        );
+        assert_eq!(
+            remote_herdr.executable.saved_bridge_command("agents"),
+            "exec \"$HOME/.local/bin/herdr\" --session agents remote-client-bridge </dev/null"
         );
     }
 
@@ -3167,8 +3692,10 @@ mod tests {
             .expect("path binary");
 
         assert_eq!(
-            remote_bridge_command(&remote_herdr, crate::session::DEFAULT_SESSION_NAME),
-            "exec /usr/bin/herdr remote-client-bridge"
+            remote_herdr
+                .executable
+                .bridge_command(crate::session::DEFAULT_SESSION_NAME),
+            "printf '\n%s\n' 'herdr-remote-output-ready:1'\nexec /usr/bin/herdr remote-client-bridge"
         );
     }
 
@@ -3183,8 +3710,10 @@ mod tests {
                 .expect("path binary");
 
         assert_eq!(
-            remote_bridge_command(&remote_herdr, crate::session::DEFAULT_SESSION_NAME),
-            "exec '/opt/herdr bin/herdr' remote-client-bridge"
+            remote_herdr
+                .executable
+                .bridge_command(crate::session::DEFAULT_SESSION_NAME),
+            "printf '\n%s\n' 'herdr-remote-output-ready:1'\nexec '/opt/herdr bin/herdr' remote-client-bridge"
         );
     }
 
@@ -3199,8 +3728,10 @@ mod tests {
                 .expect("path binary");
 
         assert_eq!(
-            remote_bridge_command(&remote_herdr, crate::session::DEFAULT_SESSION_NAME),
-            "exec /opt/homebrew/bin/herdr remote-client-bridge"
+            remote_herdr
+                .executable
+                .bridge_command(crate::session::DEFAULT_SESSION_NAME),
+            "printf '\n%s\n' 'herdr-remote-output-ready:1'\nexec /opt/homebrew/bin/herdr remote-client-bridge"
         );
         assert_eq!(remote_herdr.platform.asset_key(), "macos-aarch64");
     }
@@ -3217,8 +3748,14 @@ mod tests {
         );
 
         assert_eq!(candidates.len(), 2);
-        assert_eq!(candidates[0].shell_path, "/usr/bin/herdr");
-        assert_eq!(candidates[1].shell_path, "'/opt/herdr bin/herdr'");
+        assert_eq!(
+            candidates[0].executable,
+            RemoteExecutable::PosixShellPath("/usr/bin/herdr".to_string())
+        );
+        assert_eq!(
+            candidates[1].executable,
+            RemoteExecutable::PosixShellPath("'/opt/herdr bin/herdr'".to_string())
+        );
     }
 
     #[test]
@@ -3234,8 +3771,10 @@ mod tests {
 
         assert_eq!(candidates.len(), 1);
         assert_eq!(
-            candidates[0].shell_path,
-            "/home/can/.local/share/mise/installs/herdr/0.7.1/bin/herdr"
+            candidates[0].executable,
+            RemoteExecutable::PosixShellPath(
+                "/home/can/.local/share/mise/installs/herdr/0.7.1/bin/herdr".to_string()
+            )
         );
     }
 
@@ -3286,8 +3825,10 @@ mod tests {
                 .expect("path binary");
 
         assert_eq!(
-            remote_bridge_command(&remote_herdr, crate::session::DEFAULT_SESSION_NAME),
-            "exec '/opt/herdr'\\''s/bin/herdr' remote-client-bridge"
+            remote_herdr
+                .executable
+                .bridge_command(crate::session::DEFAULT_SESSION_NAME),
+            "printf '\n%s\n' 'herdr-remote-output-ready:1'\nexec '/opt/herdr'\\''s/bin/herdr' remote-client-bridge"
         );
     }
 
@@ -3612,8 +4153,7 @@ mod tests {
             os: "linux",
             arch: "x86_64",
         });
-        let command = remote_live_handoff_command(
-            &remote_herdr,
+        let command = remote_herdr.executable.live_handoff_command(
             crate::session::DEFAULT_SESSION_NAME,
             19,
             "0.7.9",
diff --git a/src/remote/host_unix.rs b/src/remote/host.rs
index 48b20e0b..1ff6d1e5 100644
--- a/src/remote/host_unix.rs
+++ b/src/remote/host.rs
@@ -1,15 +1,13 @@
-//! Unix remote-host side of the SSH stdio bridge.
+//! Remote-host side of the SSH stdio bridge.
 
 use std::io;
-use std::os::unix::net::UnixStream;
-use std::thread;
 use std::time::Duration;
 
 pub(crate) fn run_remote_client_bridge() -> io::Result<()> {
     ensure_remote_server_running()?;
 
     let socket_path = crate::server::socket_paths::client_socket_path();
-    let stream = UnixStream::connect(&socket_path).map_err(|err| {
+    let stream = crate::ipc::connect_local_stream(&socket_path).map_err(|err| {
         io::Error::new(
             err.kind(),
             format!(
@@ -19,33 +17,7 @@ pub(crate) fn run_remote_client_bridge() -> io::Result<()> {
         )
     })?;
 
-    let mut stdout = io::stdout().lock();
-    let mut socket_to_stdout = stream.try_clone()?;
-    let mut stdin_to_socket = stream;
-
-    let _upload = thread::spawn(move || {
-        let mut stdin = io::stdin();
-        let _ = copy_flush(&mut stdin, &mut stdin_to_socket);
-        let _ = stdin_to_socket.shutdown(std::net::Shutdown::Write);
-    });
-
-    copy_flush(&mut socket_to_stdout, &mut stdout).map(|_| ())
-}
-
-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 read = match reader.read(&mut buffer) {
-            Ok(0) => return Ok(total),
-            Ok(read) => read,
-            Err(err) if err.kind() == io::ErrorKind::Interrupted => continue,
-            Err(err) => return Err(err),
-        };
-        writer.write_all(&buffer[..read])?;
-        writer.flush()?;
-        total += read as u64;
-    }
+    crate::platform::forward_remote_bridge_stdio(stream)
 }
 
 fn ensure_remote_server_running() -> io::Result<()> {