trycua · f-trycua · Jul 1, 2026 · Jun 26, 2026
@@ -30,6 +30,7 @@ pub mod recording_zoom;
 pub mod server;
 pub mod session;
 pub mod session_tools;
+pub mod socket_io;
 pub mod text_sanitize;
 pub mod tool;
 pub mod tool_args;

@@ -0,0 +1,135 @@
+//! Reliable framed write to the cua-driver daemon socket.
+//!
+//! Split out of `serve.rs::send_request` so its EAGAIN-retry logic is unit
+//! testable without linking the platform crates (and their Swift/Metal interop)
+//! that the `cua-driver` binary pulls in.
+
+use std::io::Write;
+use std::time::Instant;
+
+/// Write `bytes` in full to a daemon socket that has `SO_SNDTIMEO` set,
+/// treating a write timeout (`WouldBlock`/`TimedOut`, i.e. EAGAIN) as "the
+/// daemon is still draining, keep waiting" rather than a fatal transport error.
+///
+/// This is the write-side mirror of `send_request`'s read loop (#1997 for
+/// #1864): a daemon momentarily too busy to read our request is not a transport
+/// failure, just as a daemon still computing a slow response is not. Without it,
+/// a single 5s `SO_SNDTIMEO` write timeout surfaced as a fatal `daemon transport
+/// error forwarding '<tool>': Resource temporarily unavailable (os error 35)`
+/// even for a tiny request. Bounded by `deadline` so a genuinely stuck daemon
+/// still surfaces an error instead of blocking forever.
+pub fn write_all_with_retry<W: Write>(
+    w: &mut W,
+    bytes: &[u8],
+    deadline: Instant,
+) -> std::io::Result<()> {
+    let mut written = 0;
+    while written < bytes.len() {
+        match w.write(&bytes[written..]) {
+            Ok(0) => {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::WriteZero,
+                    "daemon closed the connection mid-request",
+                ));
+            }
+            Ok(n) => written += n,
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            // Write timeout (SO_SNDTIMEO) / non-blocking EAGAIN: the daemon is
+            // momentarily not reading. Keep waiting until the deadline rather
+            // than surfacing a fatal transport error.
+            Err(e)
+                if matches!(
+                    e.kind(),
+                    std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut
+                ) =>
+            {
+                if Instant::now() >= deadline {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::TimedOut,
+                        format!(
+                            "daemon did not drain the socket in time \
+                             (wrote {written}/{} bytes)",
+                            bytes.len()
+                        ),
+                    ));
+                }
+                continue;
+            }
+            Err(e) => return Err(e),
+        }
+    }
+    w.flush()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::write_all_with_retry;
+    use std::io::{Error, ErrorKind, Write};
+    use std::time::{Duration, Instant};
+
+    /// Returns `WouldBlock` (EAGAIN) for its first `eagain_left` write attempts,
+    /// then accepts data — models a daemon briefly too busy to drain the socket
+    /// (the backpressure that triggers the bug).
+    struct FlakyWriter {
+        eagain_left: usize,
+        written: Vec<u8>,
+    }
+    impl Write for FlakyWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            if self.eagain_left > 0 {
+                self.eagain_left -= 1;
+                return Err(Error::new(ErrorKind::WouldBlock, "eagain"));
+            }
+            self.written.extend_from_slice(buf);
+            Ok(buf.len())
+        }
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn retries_through_transient_eagain() {
+        let mut w = FlakyWriter { eagain_left: 3, written: vec![] };
+        let deadline = Instant::now() + Duration::from_secs(5);
+        write_all_with_retry(&mut w, b"hello\n", deadline)
+            .expect("transient EAGAIN should be retried, not fatal");
+        assert_eq!(w.written, b"hello\n");
+    }
+
+    #[test]
+    fn times_out_when_daemon_never_drains() {
+        let mut w = FlakyWriter { eagain_left: usize::MAX, written: vec![] };
+        let deadline = Instant::now() + Duration::from_millis(30);
+        let err = write_all_with_retry(&mut w, b"hello\n", deadline)
+            .expect_err("a never-draining daemon must eventually surface an error");
+        assert_eq!(
+            err.kind(),
+            ErrorKind::TimedOut,
+            "deadline breach should report TimedOut, not a bare EAGAIN"
+        );
+    }
+
+    #[test]
+    fn accumulates_partial_writes() {
+        // Accepts only 2 bytes per call — exercises the offset bookkeeping so a
+        // short write keeps going until every byte lands.
+        struct ChunkWriter {
+            written: Vec<u8>,
+        }
+        impl Write for ChunkWriter {
+            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+                let n = buf.len().min(2);
+                self.written.extend_from_slice(&buf[..n]);
+                Ok(n)
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        let mut w = ChunkWriter { written: vec![] };
+        let deadline = Instant::now() + Duration::from_secs(5);
+        write_all_with_retry(&mut w, b"abcdefg\n", deadline).unwrap();
+        assert_eq!(w.written, b"abcdefg\n");
+    }
+}
@@ -358,7 +358,7 @@ pub fn read_pid_file(pid_file_path: &str) -> Option<u32> {
 /// Uses a 3-second connect timeout (by polling) and a 10-second read timeout.
 #[cfg(unix)]
 pub fn send_request(socket_path: &str, req: &DaemonRequest) -> anyhow::Result<DaemonResponse> {
-    use std::io::{Read, Write};
+    use std::io::Read;
     use std::os::unix::net::UnixStream;
     use std::time::{Duration, Instant};
 
@@ -372,8 +372,14 @@ pub fn send_request(socket_path: &str, req: &DaemonRequest) -> anyhow::Result<Da
 
     let mut w = stream.try_clone()?;
     let line = serde_json::to_string(req)? + "\n";
-    w.write_all(line.as_bytes())?;
-    w.flush()?;
+    // EAGAIN-aware write: a daemon momentarily too busy to read our request
+    // (backpressure under concurrent slow tools) makes the 5s SO_SNDTIMEO write
+    // time out. Treat that like the read loop below does (#1997) — keep retrying
+    // until an overall deadline — instead of failing with a fatal "Resource
+    // temporarily unavailable (os error 35)" transport error. Mirrors the 120s
+    // read budget.
+    let write_deadline = Instant::now() + Duration::from_secs(120);
+    cua_driver_core::socket_io::write_all_with_retry(&mut w, line.as_bytes(), write_deadline)?;
 
     // Read the single newline-terminated response line. A blocking UnixStream
     // with SO_RCVTIMEO returns `WouldBlock`/`TimedOut` (EAGAIN, os error 35)