1 files changed, 39 insertions, 21 deletions

diff --git a/src/api/server.rs b/src/api/server.rs
index e2c4b4a1..94e8c725 100644
--- a/src/api/server.rs
+++ b/src/api/server.rs
@@ -1,13 +1,13 @@
 use std::io::{self, Read, Write};
-use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
+use interprocess::local_socket::traits::{ListenerExt as _, Stream as _};
 use tracing::{debug, error, info, warn};
 
-#[cfg(test)]
+#[cfg(all(test, unix))]
 use std::fs;
 
 use crate::api::schema::{
@@ -16,7 +16,10 @@ use crate::api::schema::{
 use crate::api::subscriptions::ActiveSubscription;
 use crate::api::wait::wait_for_output;
 use crate::api::{request_changes_ui, socket_path, ApiRequestMessage, ApiRequestSender, EventHub};
-use crate::ipc::{remove_socket_file_if_owned, socket_file_identity, SocketFileIdentity};
+use crate::ipc::{
+    bind_local_listener, remove_socket_file_if_owned, socket_file_identity, LocalStream,
+    SocketFileIdentity,
+};
 
 const SOCKET_PERMISSION_MODE: u32 = 0o600;
 pub(super) const CONNECTION_POLL_INTERVAL: Duration = Duration::from_millis(100);
@@ -46,7 +49,7 @@ impl Drop for ServerHandle {
 
 impl ServerHandle {
     pub(crate) fn remove_socket_file_if_owned(&self) -> std::io::Result<()> {
-        remove_socket_file_if_owned(&self.path, self.identity)
+        remove_socket_file_if_owned(&self.path, &self.identity)
     }
 }
 
@@ -57,7 +60,9 @@ pub fn start_server(
     start_server_with_capabilities(
         api_tx,
         event_hub,
-        Some(ServerCapabilities { live_handoff: true }),
+        Some(ServerCapabilities {
+            live_handoff: crate::platform::capabilities().live_handoff,
+        }),
     )
 }
 
@@ -69,7 +74,7 @@ pub fn start_server_with_capabilities(
     let path = socket_path();
     prepare_socket_path(&path)?;
 
-    let listener = UnixListener::bind(&path)?;
+    let listener = bind_local_listener(&path)?;
     restrict_socket_permissions(&path)?;
     let identity = socket_file_identity(&path)?;
     info!(path = %path.display(), "api server listening");
@@ -127,13 +132,13 @@ fn restrict_socket_permissions(path: &Path) -> std::io::Result<()> {
 }
 
 fn handle_connection(
-    mut stream: UnixStream,
+    mut stream: LocalStream,
     api_tx: &ApiRequestSender,
     event_hub: &EventHub,
     running: &Arc<AtomicBool>,
     capabilities: Option<ServerCapabilities>,
 ) -> std::io::Result<()> {
-    if let Err(err) = stream.set_write_timeout(Some(STREAM_WRITE_TIMEOUT)) {
+    if let Err(err) = stream.set_send_timeout(Some(STREAM_WRITE_TIMEOUT)) {
         debug!(err = %err, "api connection write timeout unavailable");
     }
 
@@ -340,7 +345,7 @@ fn api_response_outcome(response: &str) -> &'static str {
     }
 }
 
-fn read_initial_request_line(stream: &mut UnixStream) -> std::io::Result<Option<String>> {
+fn read_initial_request_line(stream: &mut LocalStream) -> std::io::Result<Option<String>> {
     stream.set_nonblocking(true)?;
     let deadline = Instant::now() + INITIAL_REQUEST_TIMEOUT;
     let mut bytes = Vec::new();
@@ -387,7 +392,7 @@ fn read_initial_request_line(stream: &mut UnixStream) -> std::io::Result<Option<
 }
 
 fn stream_subscriptions(
-    mut stream: UnixStream,
+    mut stream: LocalStream,
     request_id: String,
     params: crate::api::schema::EventsSubscribeParams,
     api_tx: &ApiRequestSender,
@@ -444,27 +449,30 @@ fn stream_subscriptions(
     }
 }
 
-fn write_text_line(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
+fn write_text_line(stream: &mut LocalStream, value: &str) -> std::io::Result<()> {
     stream.write_all(value.as_bytes())?;
     stream.write_all(b"\n")?;
     stream.flush()
 }
 
-fn write_text_line_allow_disconnect(stream: &mut UnixStream, value: &str) -> std::io::Result<()> {
+fn write_text_line_allow_disconnect(stream: &mut LocalStream, value: &str) -> std::io::Result<()> {
     match write_text_line(stream, value) {
         Err(err) if is_connection_closed_error(&err) => Ok(()),
         result => result,
     }
 }
 
-fn write_json_line<T: serde::Serialize>(stream: &mut UnixStream, value: &T) -> std::io::Result<()> {
+fn write_json_line<T: serde::Serialize>(
+    stream: &mut LocalStream,
+    value: &T,
+) -> std::io::Result<()> {
     let encoded = serde_json::to_string(value)
         .map_err(|err| std::io::Error::other(format!("failed to encode json: {err}")))?;
     write_text_line(stream, &encoded)
 }
 
 fn write_json_line_allow_disconnect<T: serde::Serialize>(
-    stream: &mut UnixStream,
+    stream: &mut LocalStream,
     value: &T,
 ) -> std::io::Result<()> {
     let encoded = serde_json::to_string(value)
@@ -473,7 +481,7 @@ fn write_json_line_allow_disconnect<T: serde::Serialize>(
 }
 
 pub(super) fn should_stop_connection(
-    stream: &mut UnixStream,
+    stream: &mut LocalStream,
     running: &Arc<AtomicBool>,
 ) -> std::io::Result<bool> {
     if !running.load(Ordering::Relaxed) {
@@ -483,7 +491,7 @@ pub(super) fn should_stop_connection(
     probe_stream_closed(stream)
 }
 
-fn probe_stream_closed(stream: &mut UnixStream) -> std::io::Result<bool> {
+fn probe_stream_closed(stream: &mut LocalStream) -> std::io::Result<bool> {
     stream.set_nonblocking(true)?;
     let mut probe = [0u8; 1];
     let status = match stream.read(&mut probe) {
@@ -581,11 +589,13 @@ fn error_response_json(id: String, code: &str, message: String) -> String {
     })
 }
 
-#[cfg(test)]
+#[cfg(all(test, unix))]
 mod tests {
     use super::*;
+    use interprocess::local_socket::traits::Listener as _;
     use std::io::{BufRead, BufReader};
     use std::os::unix::fs::PermissionsExt;
+    use std::os::unix::net::UnixListener;
     use std::sync::{Mutex, OnceLock};
     use tokio::sync::mpsc;
 
@@ -602,13 +612,21 @@ mod tests {
         std::env::temp_dir().join(format!("herdr-{name}-{}-{nanos}", std::process::id()))
     }
 
-    fn read_line(stream: &mut UnixStream) -> String {
+    fn read_line(stream: &mut LocalStream) -> String {
         let mut reader = BufReader::new(stream);
         let mut line = String::new();
         reader.read_line(&mut line).unwrap();
         line
     }
 
+    fn local_stream_pair(name: &str) -> (LocalStream, LocalStream, PathBuf) {
+        let path = unique_test_path(name);
+        let listener = crate::ipc::bind_local_listener(&path).unwrap();
+        let client = crate::ipc::connect_local_stream(&path).unwrap();
+        let server = listener.accept().unwrap();
+        (client, server, path)
+    }
+
     #[test]
     fn socket_path_prefers_explicit_env_override() {
         let _guard = env_lock().lock().unwrap();
@@ -770,7 +788,7 @@ mod tests {
             }
         });
 
-        let (mut client, server) = UnixStream::pair().unwrap();
+        let (mut client, server, _path) = local_stream_pair("api-wait-disconnect");
         client
             .write_all(br#"{"id":"req_wait","method":"pane.wait_for_output","params":{"pane_id":"pane_1","source":"recent","match":{"type":"substring","value":"never"}}}"#)
             .unwrap();
@@ -800,7 +818,7 @@ mod tests {
     #[test]
     fn subscriptions_stop_when_client_disconnects() {
         let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server) = UnixStream::pair().unwrap();
+        let (mut client, server, _path) = local_stream_pair("api-sub-disconnect");
         client
             .write_all(
                 br#"{"id":"sub_1","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,
@@ -832,7 +850,7 @@ mod tests {
     #[test]
     fn subscriptions_stop_when_server_shuts_down() {
         let (api_tx, _api_rx) = mpsc::unbounded_channel::<ApiRequestMessage>();
-        let (mut client, server) = UnixStream::pair().unwrap();
+        let (mut client, server, _path) = local_stream_pair("api-sub-shutdown");
         client
             .write_all(
                 br#"{"id":"sub_2","method":"events.subscribe","params":{"subscriptions":[{"type":"workspace.created"}]}}"#,