fix: clean up failed terminal observer setup (#4071)

refs #3612

4 files changed, 23 insertions, 9 deletions

diff --git a/src/platform/linux.rs b/src/platform/linux.rs
index 23393653..001d0798 100644
--- a/src/platform/linux.rs
+++ b/src/platform/linux.rs
@@ -16,8 +16,9 @@ pub(crate) use super::unix_common::{
     configure_status_command, create_remote_private_dir, create_remote_ssh_config_dir,
     create_remote_ssh_config_file, hostname, local_datetime, remote_bridge_endpoint_path,
     remote_private_temp_base, remote_reattach_argument, remote_reattach_program,
-    remote_ssh_config_paths, set_default_plugin_pane_pwd, status_commands_supported,
-    wait_client_stream_readable, write_client_stream, ClientStreamReader, StatusCommandGuard,
+    remote_ssh_config_paths, set_default_plugin_pane_pwd, shutdown_client_stream,
+    status_commands_supported, wait_client_stream_readable, write_client_stream,
+    ClientStreamReader, StatusCommandGuard,
 };
 
 const WSL_MARKER_ENV_VARS: &[&str] = &["WSL_DISTRO_NAME", "WSL_INTEROP"];
diff --git a/src/platform/macos.rs b/src/platform/macos.rs
index 453d3fb3..422bbfca 100644
--- a/src/platform/macos.rs
+++ b/src/platform/macos.rs
@@ -16,8 +16,9 @@ pub(crate) use super::unix_common::{
     configure_status_command, create_remote_private_dir, create_remote_ssh_config_dir,
     create_remote_ssh_config_file, hostname, local_datetime, remote_bridge_endpoint_path,
     remote_private_temp_base, remote_reattach_argument, remote_reattach_program,
-    remote_ssh_config_paths, set_default_plugin_pane_pwd, status_commands_supported,
-    wait_client_stream_readable, write_client_stream, ClientStreamReader, StatusCommandGuard,
+    remote_ssh_config_paths, set_default_plugin_pane_pwd, shutdown_client_stream,
+    status_commands_supported, wait_client_stream_readable, write_client_stream,
+    ClientStreamReader, StatusCommandGuard,
 };
 
 const PROC_PGRP_ONLY: u32 = 2;
diff --git a/src/platform/unix_common.rs b/src/platform/unix_common.rs
index ccb068f8..a633f283 100644
--- a/src/platform/unix_common.rs
+++ b/src/platform/unix_common.rs
@@ -8,6 +8,11 @@ pub(crate) fn classify_child_exit(status: &portable_pty::ExitStatus) -> super::C
     }
 }
 
+pub(crate) fn shutdown_client_stream(stream: &crate::ipc::LocalStream) -> std::io::Result<()> {
+    let crate::ipc::LocalStream::UdSocket(stream) = stream;
+    stream.inner().shutdown(std::net::Shutdown::Both)
+}
+
 pub(crate) struct ClientStreamReader<'a>(pub(crate) &'a mut crate::ipc::LocalStream);
 
 impl std::io::Read for ClientStreamReader<'_> {
@@ -45,14 +50,14 @@ pub(crate) fn write_client_stream(
     use std::os::fd::AsRawFd as _;
     use std::time::Instant;
 
-    let crate::ipc::LocalStream::UdSocket(stream) = stream;
-    let mut socket = stream.inner();
+    let crate::ipc::LocalStream::UdSocket(socket) = stream;
+    let mut socket = socket.inner();
     let Some(timeout) = socket.write_timeout()? else {
         return socket.write_all(data);
     };
     let timed_out = || {
         // Dropping the writer clone alone would leave the reader blocked.
-        let _ = stream.inner().shutdown(std::net::Shutdown::Both);
+        let _ = shutdown_client_stream(stream);
         io::Error::new(
             io::ErrorKind::TimedOut,
             "terminal observer stopped receiving output",
diff --git a/src/server/client_transport.rs b/src/server/client_transport.rs
index da682c06..210a840b 100644
--- a/src/server/client_transport.rs
+++ b/src/server/client_transport.rs
@@ -1042,10 +1042,17 @@ fn client_read_loop_with_endpoint_controls(
             ClientMessage::ObserveTerminal { target } => {
                 #[cfg(unix)]
                 {
-                    stream.set_send_timeout(Some(OBSERVER_WRITE_TIMEOUT))?;
                     // macOS Unix sockets can block even with per-send MSG_DONTWAIT.
                     // ClientStreamReader preserves blocking reads on the shared socket.
-                    stream.set_nonblocking(true)?;
+                    let configured = stream
+                        .set_send_timeout(Some(OBSERVER_WRITE_TIMEOUT))
+                        .and_then(|()| stream.set_nonblocking(true));
+                    if let Err(err) = configured {
+                        let _ = crate::platform::shutdown_client_stream(&stream);
+                        let _ = server_event_tx
+                            .blocking_send(ServerEvent::ClientDisconnected { client_id });
+                        return Err(err);
+                    }
                 }
                 ServerEvent::ClientObserveTerminal { client_id, target }
             }