fix: propagate host color scheme to pane apps (#2214)

refs #714

2 files changed, 182 insertions, 29 deletions

diff --git a/src/pty/actor.rs b/src/pty/actor.rs
index 8c3a6b52..6fde5051 100644
--- a/src/pty/actor.rs
+++ b/src/pty/actor.rs
@@ -52,6 +52,8 @@ mod windows {
     pub(crate) struct PtyIoActorHandle {
         data_tx: mpsc::Sender<Bytes>,
         control_tx: std_mpsc::Sender<PtyIoControlCommand>,
+        write_tx: std_mpsc::Sender<Bytes>,
+        response_order: Arc<Mutex<()>>,
         accepting: Arc<Mutex<bool>>,
     }
 
@@ -84,6 +86,16 @@ mod windows {
             self.data_tx.try_send(bytes)
         }
 
+        pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
+            let _order = self
+                .response_order
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner());
+            if let Some(bytes) = response().filter(|bytes| !bytes.is_empty()) {
+                let _ = self.write_tx.send(bytes);
+            }
+        }
+
         pub(crate) fn resize(
             &self,
             rows: u16,
@@ -128,39 +140,55 @@ mod windows {
             let mut reader = master
                 .try_clone_reader()
                 .map_err(|err| std::io::Error::other(err.to_string()))?;
-            let writer = master
+            let mut writer = master
                 .take_writer()
                 .map_err(|err| std::io::Error::other(err.to_string()))?;
-            let writer = Arc::new(Mutex::new(writer));
             let (data_tx, mut data_rx) = mpsc::channel::<Bytes>(1024);
             let (control_tx, control_rx) = std_mpsc::channel::<PtyIoControlCommand>();
+            let (write_tx, write_rx) = std_mpsc::channel::<Bytes>();
+            let response_order = Arc::new(Mutex::new(()));
             let accepting = Arc::new(Mutex::new(!initially_quiesced));
 
+            std::thread::spawn(move || {
+                for bytes in write_rx {
+                    if writer.write_all(&bytes).is_err() || writer.flush().is_err() {
+                        break;
+                    }
+                }
+                debug!(pane_id, "windows pty writer thread exiting");
+            });
+
             {
-                let writer = Arc::clone(&writer);
+                let write_tx = write_tx.clone();
                 std::thread::spawn(move || {
                     while let Some(bytes) = data_rx.blocking_recv() {
-                        if write_all_locked(&writer, &bytes).is_err() {
+                        if write_tx.send(bytes).is_err() {
                             break;
                         }
                     }
-                    debug!(pane_id, "windows pty writer thread exiting");
+                    debug!(pane_id, "windows pty input thread exiting");
                 });
             }
 
             {
-                let writer = Arc::clone(&writer);
+                let write_tx = write_tx.clone();
+                let response_order = Arc::clone(&response_order);
                 std::thread::spawn(move || {
                     let mut buf = [0u8; 8192];
                     loop {
                         match reader.read(&mut buf) {
                             Ok(0) => break,
                             Ok(n) => {
+                                let _order = response_order
+                                    .lock()
+                                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                                 let result = on_read(&buf[..n]);
-                                for response in result.terminal_responses {
-                                    if write_all_locked(&writer, &response).is_err() {
-                                        break;
-                                    }
+                                if result
+                                    .terminal_responses
+                                    .into_iter()
+                                    .any(|response| write_tx.send(response).is_err())
+                                {
+                                    break;
                                 }
                             }
                             Err(err) => {
@@ -177,7 +205,7 @@ mod windows {
             }
 
             {
-                let writer = Arc::clone(&writer);
+                let write_tx = write_tx.clone();
                 std::thread::spawn(move || {
                     for command in control_rx {
                         match command {
@@ -191,10 +219,12 @@ mod windows {
                                 }) {
                                     warn!(pane_id, err = %err, "windows pty resize failed");
                                 }
-                                for response in request.terminal_responses {
-                                    if write_all_locked(&writer, &response).is_err() {
-                                        break;
-                                    }
+                                if request
+                                    .terminal_responses
+                                    .into_iter()
+                                    .any(|response| write_tx.send(response).is_err())
+                                {
+                                    break;
                                 }
                             }
                             PtyIoControlCommand::Shutdown => break,
@@ -207,22 +237,13 @@ mod windows {
             Ok(PtyIoActorHandle {
                 data_tx,
                 control_tx,
+                write_tx,
+                response_order,
                 accepting,
             })
         }
     }
 
-    fn write_all_locked(
-        writer: &Arc<Mutex<Box<dyn Write + Send>>>,
-        bytes: &[u8],
-    ) -> std::io::Result<()> {
-        let mut writer = writer
-            .lock()
-            .unwrap_or_else(|poisoned| poisoned.into_inner());
-        writer.write_all(bytes)?;
-        writer.flush()
-    }
-
     #[allow(dead_code)]
     fn _assert_duration_send(_: Duration) {}
 }
diff --git a/src/pty/actor/unix.rs b/src/pty/actor/unix.rs
index 426c3327..599f02cb 100644
--- a/src/pty/actor/unix.rs
+++ b/src/pty/actor/unix.rs
@@ -60,6 +60,7 @@ struct PtyResizeRequest {
 struct SharedPtyControls {
     resize: Option<PtyResizeRequest>,
     nudge: Option<PtyResize>,
+    terminal_responses: Vec<Bytes>,
 }
 
 pub(crate) struct PtyIoActorConfig {
@@ -90,6 +91,7 @@ pub(crate) struct PtyIoActorHandle {
     wake: fd::WakeWriter,
     user_writes: Arc<Mutex<UserWriteGate>>,
     controls: Arc<Mutex<SharedPtyControls>>,
+    response_order: Arc<Mutex<()>>,
 }
 
 #[derive(Debug)]
@@ -157,6 +159,24 @@ impl PtyIoActorHandle {
         }
     }
 
+    pub(crate) fn write_terminal_response(&self, response: impl FnOnce() -> Option<Bytes>) {
+        let _order = self
+            .response_order
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let Some(bytes) = response() else {
+            return;
+        };
+        if !bytes.is_empty() {
+            self.controls
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .terminal_responses
+                .push(bytes);
+            self.wake_actor();
+        }
+    }
+
     pub(crate) fn resize(
         &self,
         rows: u16,
@@ -349,12 +369,14 @@ impl PtyIoActor {
             accepting: !config.initially_quiesced,
         }));
         let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
+        let response_order = Arc::new(Mutex::new(()));
         let handle = PtyIoActorHandle {
             data_tx,
             control_tx,
             wake: wake_pipe.writer,
             user_writes,
             controls: Arc::clone(&controls),
+            response_order: Arc::clone(&response_order),
         };
 
         let mut runner = PtyIoActorRunner {
@@ -371,6 +393,7 @@ impl PtyIoActor {
             current_write_offset: 0,
             wake_read_fd: wake_pipe.read_fd,
             controls,
+            response_order,
             on_read: config.on_read,
             on_reader_exit: config.on_reader_exit,
             poll_observer,
@@ -402,6 +425,7 @@ struct PtyIoActorRunner {
     current_write_offset: usize,
     wake_read_fd: OwnedFd,
     controls: Arc<Mutex<SharedPtyControls>>,
+    response_order: Arc<Mutex<()>>,
     on_read: ReadCallback,
     on_reader_exit: Option<ReaderExitCallback>,
     poll_observer: Option<std_mpsc::Sender<()>>,
@@ -625,12 +649,16 @@ impl PtyIoActorRunner {
     }
 
     fn apply_pending_controls(&mut self) {
-        let (resize, nudge) = {
+        let (resize, nudge, terminal_responses) = {
             let mut controls = self
                 .controls
                 .lock()
                 .unwrap_or_else(|poisoned| poisoned.into_inner());
-            (controls.resize.take(), controls.nudge.take())
+            (
+                controls.resize.take(),
+                controls.nudge.take(),
+                std::mem::take(&mut controls.terminal_responses),
+            )
         };
         if self.state == ActorState::Released {
             return;
@@ -642,6 +670,7 @@ impl PtyIoActorRunner {
         if let Some(nudge) = nudge {
             self.nudge(nudge);
         }
+        self.enqueue_terminal_responses(terminal_responses);
     }
 
     fn read_once(&mut self) -> bool {
@@ -655,8 +684,25 @@ impl PtyIoActorRunner {
                 false
             }
             Ok(n) => {
+                let response_order = Arc::clone(&self.response_order);
+                let _order = response_order
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner());
                 let result = (self.on_read)(&buf[..n]);
-                self.enqueue_terminal_responses(result.terminal_responses);
+                self.controls
+                    .lock()
+                    .unwrap_or_else(|poisoned| poisoned.into_inner())
+                    .terminal_responses
+                    .extend(result.terminal_responses);
+                drop(_order);
+                let terminal_responses = std::mem::take(
+                    &mut self
+                        .controls
+                        .lock()
+                        .unwrap_or_else(|poisoned| poisoned.into_inner())
+                        .terminal_responses,
+                );
+                self.enqueue_terminal_responses(terminal_responses);
                 true
             }
         }
@@ -769,6 +815,7 @@ mod tests {
         io::{Read, Write},
         os::fd::{AsRawFd, FromRawFd, IntoRawFd},
         os::unix::net::UnixStream,
+        sync::atomic::{AtomicBool, Ordering},
     };
 
     fn test_wake_pair() -> (fd::WakeWriter, OwnedFd) {
@@ -834,6 +881,7 @@ mod tests {
             current_write_offset: 0,
             wake_read_fd: wake_pipe.read_fd,
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
+            response_order: Arc::new(Mutex::new(())),
             on_read: Box::new(|_| PtyReadResult::empty()),
             on_reader_exit: None,
             poll_observer: None,
@@ -1099,11 +1147,13 @@ mod tests {
             wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::clone(&controls),
+            response_order: Arc::new(Mutex::new(())),
         };
 
         handle.resize(20, 80, 8, 16, vec![Bytes::from_static(b"old")]);
         handle.resize(40, 120, 9, 18, vec![Bytes::from_static(b"new")]);
         handle.nudge_child_redraw_after_handoff(41, 121, 10, 20);
+        handle.write_terminal_response(|| Some(Bytes::from_static(b"response")));
 
         let controls = controls.lock().expect("controls lock");
         assert_eq!(
@@ -1127,6 +1177,84 @@ mod tests {
                 cell_height_px: 20,
             })
         );
+        assert_eq!(
+            controls.terminal_responses,
+            vec![Bytes::from_static(b"response")]
+        );
+    }
+
+    #[test]
+    fn appearance_transition_report_precedes_query_of_new_scheme() {
+        let (actor_socket, mut peer) = UnixStream::pair().expect("socket pair");
+        actor_socket
+            .set_nonblocking(true)
+            .expect("actor socket nonblocking");
+        let owned = unsafe { OwnedFd::from_raw_fd(actor_socket.into_raw_fd()) };
+        let (data_tx, data_rx) = mpsc::channel(ACTOR_COMMAND_BUFFER);
+        let (control_tx, control_rx) = std_mpsc::channel();
+        let wake_pipe = fd::create_wake_pipe().expect("wake pipe");
+        let controls = Arc::new(Mutex::new(SharedPtyControls::default()));
+        let response_order = Arc::new(Mutex::new(()));
+        let light = Arc::new(AtomicBool::new(false));
+        let query_light = Arc::clone(&light);
+        let runner = PtyIoActorRunner {
+            pane_id: 1,
+            file: std::fs::File::from(owned),
+            data_rx,
+            control_rx,
+            state: ActorState::Running,
+            pending_writes: VecDeque::new(),
+            current_write_offset: 0,
+            wake_read_fd: wake_pipe.read_fd,
+            controls: Arc::clone(&controls),
+            response_order: Arc::clone(&response_order),
+            on_read: Box::new(move |_| PtyReadResult {
+                terminal_responses: vec![if query_light.load(Ordering::Acquire) {
+                    Bytes::from_static(b"query-light")
+                } else {
+                    Bytes::from_static(b"query-dark")
+                }],
+            }),
+            on_reader_exit: None,
+            poll_observer: None,
+        };
+        let handle = PtyIoActorHandle {
+            data_tx,
+            control_tx,
+            wake: wake_pipe.writer,
+            user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
+            controls,
+            response_order,
+        };
+        let (changed_tx, changed_rx) = std_mpsc::channel();
+        let (continue_tx, continue_rx) = std_mpsc::channel();
+
+        let appearance = std::thread::spawn(move || {
+            handle.write_terminal_response(|| {
+                light.store(true, Ordering::Release);
+                changed_tx.send(()).expect("notify appearance change");
+                continue_rx.recv().expect("continue appearance report");
+                Some(Bytes::from_static(b"live-light"))
+            });
+        });
+        changed_rx.recv().expect("appearance changed");
+        peer.write_all(b"query").expect("write query");
+        let reader = std::thread::spawn(move || {
+            let mut runner = runner;
+            assert!(runner.read_once());
+            runner
+        });
+        continue_tx.send(()).expect("release appearance report");
+        appearance.join().expect("appearance thread joins");
+        let runner = reader.join().expect("reader thread joins");
+
+        assert_eq!(
+            runner.pending_writes,
+            VecDeque::from([
+                Bytes::from_static(b"live-light"),
+                Bytes::from_static(b"query-light"),
+            ])
+        );
     }
 
     #[test]
@@ -1159,6 +1287,7 @@ mod tests {
             wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
+            response_order: Arc::new(Mutex::new(())),
         };
 
         let write = tokio::spawn(async move {
@@ -1204,6 +1333,7 @@ mod tests {
             wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
+            response_order: Arc::new(Mutex::new(())),
         };
         let write_handle = handle.clone();
         let write = tokio::spawn(async move {
@@ -1258,6 +1388,7 @@ mod tests {
             wake,
             user_writes: Arc::new(Mutex::new(UserWriteGate { accepting: true })),
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
+            response_order: Arc::new(Mutex::new(())),
         };
 
         let handoff = std::thread::spawn(move || handle.begin_handoff(Duration::from_secs(1)));
@@ -1302,6 +1433,7 @@ mod tests {
             current_write_offset: 0,
             wake_read_fd: fd::create_wake_pipe().expect("wake pipe").read_fd,
             controls: Arc::new(Mutex::new(SharedPtyControls::default())),
+            response_order: Arc::new(Mutex::new(())),
             on_read: Box::new(|_| PtyReadResult::empty()),
             on_reader_exit: None,
             poll_observer: None,