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>

4 files changed, 127 insertions, 32 deletions

diff --git a/src/platform/fallback.rs b/src/platform/fallback.rs
index 09d82c6c..a533796f 100644
--- a/src/platform/fallback.rs
+++ b/src/platform/fallback.rs
@@ -13,6 +13,19 @@ pub(crate) fn set_default_plugin_pane_pwd(
 ) {
 }
 
+pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+    use interprocess::TryClone as _;
+
+    let mut stdout = std::io::stdout().lock();
+    let mut socket_to_stdout = stream.try_clone()?;
+    let mut stdin_to_socket = stream;
+    let _upload = std::thread::spawn(move || {
+        let mut stdin = std::io::stdin();
+        let _ = std::io::copy(&mut stdin, &mut stdin_to_socket);
+    });
+    std::io::copy(&mut socket_to_stdout, &mut stdout).map(|_| ())
+}
+
 #[cfg(not(unix))]
 pub(super) fn read_terminal_grid_size() -> std::io::Result<(u16, u16)> {
     crossterm::terminal::size()
diff --git a/src/platform/mod.rs b/src/platform/mod.rs
index 335293e8..67e474ff 100644
--- a/src/platform/mod.rs
+++ b/src/platform/mod.rs
@@ -264,7 +264,9 @@ pub(crate) struct RemoteSshConfigPaths {
 #[cfg(unix)]
 mod unix_common;
 #[cfg(unix)]
-pub(crate) use unix_common::{begin_cli_output, end_cli_output, RemoteBridgeWake};
+pub(crate) use unix_common::{
+    begin_cli_output, end_cli_output, forward_remote_bridge_stdio, RemoteBridgeWake,
+};
 
 mod client_state;
 pub(crate) use client_state::{create_private_state_file, replace_file, sync_parent_directory};
@@ -359,6 +361,35 @@ pub(crate) fn quote_powershell_arg(value: &str) -> String {
     format!("'{}'", value.replace('\'', "''"))
 }
 
+pub(crate) fn quote_windows_command_line_arg(value: &str) -> String {
+    if !value.is_empty()
+        && !value
+            .chars()
+            .any(|ch| matches!(ch, ' ' | '\t' | '\n' | '\x0b' | '"'))
+    {
+        return value.to_string();
+    }
+
+    let mut quoted = String::from("\"");
+    let mut backslashes = 0;
+    for ch in value.chars() {
+        if ch == '\\' {
+            backslashes += 1;
+            continue;
+        }
+        if ch == '"' {
+            quoted.push_str(&"\\".repeat(backslashes * 2 + 1));
+        } else {
+            quoted.push_str(&"\\".repeat(backslashes));
+        }
+        backslashes = 0;
+        quoted.push(ch);
+    }
+    quoted.push_str(&"\\".repeat(backslashes * 2));
+    quoted.push('"');
+    quoted
+}
+
 pub(crate) fn is_pane_shell_process_name(name: &str) -> bool {
     let normalized = normalized_process_name(name);
     matches!(
diff --git a/src/platform/unix_common.rs b/src/platform/unix_common.rs
index 0df215d6..756abc17 100644
--- a/src/platform/unix_common.rs
+++ b/src/platform/unix_common.rs
@@ -27,6 +27,38 @@ pub(crate) fn wait_client_stream_readable(stream: &crate::ipc::LocalStream) -> s
     Ok(())
 }
 
+pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+    use interprocess::TryClone as _;
+
+    let mut stdout = std::io::stdout().lock();
+    let mut socket_to_stdout = stream.try_clone()?;
+    let mut stdin_to_socket = stream;
+    let _upload = std::thread::spawn(move || {
+        let mut stdin = std::io::stdin();
+        let _ = copy_flush(&mut stdin, &mut stdin_to_socket);
+        let crate::ipc::LocalStream::UdSocket(stream) = stdin_to_socket;
+        let _ = stream.inner().shutdown(std::net::Shutdown::Write);
+    });
+    copy_flush(&mut socket_to_stdout, &mut stdout)
+}
+
+fn copy_flush<R: std::io::Read, W: std::io::Write>(
+    reader: &mut R,
+    writer: &mut W,
+) -> std::io::Result<()> {
+    let mut buffer = [0_u8; 16 * 1024];
+    loop {
+        let read = match reader.read(&mut buffer) {
+            Ok(0) => return Ok(()),
+            Ok(read) => read,
+            Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(err) => return Err(err),
+        };
+        writer.write_all(&buffer[..read])?;
+        writer.flush()?;
+    }
+}
+
 pub(crate) struct RemoteBridgeWake {
     reader: std::os::unix::net::UnixStream,
     writer: std::os::unix::net::UnixStream,
diff --git a/src/platform/windows.rs b/src/platform/windows.rs
index e58bacae..87945daf 100644
--- a/src/platform/windows.rs
+++ b/src/platform/windows.rs
@@ -52,6 +52,54 @@ pub(crate) fn wait_client_stream_readable(
     Ok(())
 }
 
+pub(crate) fn forward_remote_bridge_stdio(stream: crate::ipc::LocalStream) -> std::io::Result<()> {
+    use interprocess::TryClone as _;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    let mut stdout = std::io::stdout().lock();
+    let mut socket_to_stdout = stream.try_clone()?;
+    let mut stdin_to_socket = stream;
+    let upload_done = Arc::new(AtomicBool::new(false));
+    let upload_done_worker = Arc::clone(&upload_done);
+    let _upload = std::thread::spawn(move || {
+        let mut stdin = std::io::stdin();
+        let _ = copy_flush(&mut stdin, &mut stdin_to_socket);
+        upload_done_worker.store(true, Ordering::Release);
+    });
+
+    let mut buffer = [0_u8; 16 * 1024];
+    while !upload_done.load(Ordering::Acquire) {
+        match crate::ipc::poll_local_stream_read_count(&mut socket_to_stdout, &mut buffer)? {
+            crate::ipc::LocalStreamReadCount::Data(read) => {
+                std::io::Write::write_all(&mut stdout, &buffer[..read])?;
+                std::io::Write::flush(&mut stdout)?;
+            }
+            crate::ipc::LocalStreamReadCount::Pending => {
+                std::thread::sleep(Duration::from_millis(1));
+            }
+            crate::ipc::LocalStreamReadCount::Closed => break,
+        }
+    }
+    Ok(())
+}
+
+fn copy_flush<R: std::io::Read, W: std::io::Write>(
+    reader: &mut R,
+    writer: &mut W,
+) -> std::io::Result<()> {
+    let mut buffer = [0_u8; 16 * 1024];
+    loop {
+        let read = match reader.read(&mut buffer) {
+            Ok(0) => return Ok(()),
+            Ok(read) => read,
+            Err(err) if err.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(err) => return Err(err),
+        };
+        writer.write_all(&buffer[..read])?;
+        writer.flush()?;
+    }
+}
+
 pub(super) fn read_terminal_grid_size() -> std::io::Result<(u16, u16)> {
     crossterm::terminal::size()
 }
@@ -669,7 +717,7 @@ fn powershell_agent_script(argv: &[String]) -> Option<String> {
         .join(" ");
     let command_line = args
         .iter()
-        .map(|arg| quote_windows_command_line_arg(arg))
+        .map(|arg| super::quote_windows_command_line_arg(arg))
         .collect::<Vec<_>>()
         .join(" ");
     Some(format!(
@@ -682,35 +730,6 @@ fn powershell_agent_script(argv: &[String]) -> Option<String> {
     ))
 }
 
-fn quote_windows_command_line_arg(value: &str) -> String {
-    if !value.is_empty()
-        && !value
-            .chars()
-            .any(|ch| matches!(ch, ' ' | '\t' | '\n' | '\x0b' | '"'))
-    {
-        return value.to_string();
-    }
-
-    let mut quoted = String::from("\"");
-    let mut backslashes = 0;
-    for ch in value.chars() {
-        if ch == '\\' {
-            backslashes += 1;
-            continue;
-        }
-        if ch == '"' {
-            quoted.push_str(&"\\".repeat(backslashes * 2 + 1));
-        } else {
-            quoted.push_str(&"\\".repeat(backslashes));
-        }
-        backslashes = 0;
-        quoted.push(ch);
-    }
-    quoted.push_str(&"\\".repeat(backslashes * 2));
-    quoted.push('"');
-    quoted
-}
-
 fn cmd_encoded_powershell_command(script: &str) -> String {
     use base64::Engine as _;
 
@@ -1015,7 +1034,7 @@ fn windows_command_line(command: &std::process::Command) -> std::io::Result<Stri
         .chain(command.get_args())
         .map(|value| {
             unicode_windows_value(value, "server command argument")
-                .map(|value| quote_windows_command_line_arg(&value))
+                .map(|value| super::quote_windows_command_line_arg(&value))
         })
         .collect::<std::io::Result<Vec<_>>>()
         .map(|parts| parts.join(" "))