Implement PtyWrite event

crates/embers-server/src/buffer_runtime.rs

@@ -13,7 +13,7 @@ use std::os::windows::ffi::OsStrExt;
 use std::path::{Path, PathBuf};
 use std::process::{ChildStdin, Command as ProcessCommand, Stdio};
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::mpsc::{SyncSender, TrySendError, sync_channel};
+use std::sync::mpsc::{Receiver, SyncSender, TrySendError, sync_channel};
 use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::Duration;
@@ -27,7 +27,7 @@ use portable_pty::{
     PtySystem,
 };
 use serde::{Deserialize, Serialize};
-use tracing::error;
+use tracing::{debug, error};
 
 use crate::{AlacrittyTerminalBackend, RawByteRouter, TerminalBackend};
 
@@ -36,6 +36,16 @@ const CONNECT_RETRY_ATTEMPTS: usize = 1200;
 const STATUS_POLL_INTERVAL: Duration = Duration::from_millis(50);
 const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024;
 const KEEPER_PIPE_WRITE_QUEUE_CAPACITY: usize = 64;
+/// Bound on bytes queued for the PTY master. Every writer — client input
+/// (`KeeperRequest::Write`) and emulator replies (device-attribute /
+/// cursor-position / mode-query responses) — hands off through this queue, which
+/// a single dedicated writer thread drains. Because no caller holds a lock during
+/// the blocking `write_all`, a child that stops reading its input backpressures
+/// the queue instead of stalling request handling or the read loop. The read loop
+/// (the only thread draining child output) must never block, so it drops replies
+/// best-effort when the queue is full and keeps draining, which lets the child
+/// make progress and unblocks the writer.
+const KEEPER_WRITE_QUEUE_CAPACITY: usize = 64;
 
 #[derive(Clone, Debug)]
 pub struct BufferRuntimeUpdate {
@@ -211,7 +221,11 @@ impl KeeperScrollbackSlice {
 struct KeeperRuntime {
     surface: Mutex<KeeperSurface>,
     master: Mutex<Box<dyn MasterPty + Send>>,
-    writer: Mutex<Box<dyn Write + Send>>,
+    /// All PTY-master writes are submitted here and serialized by the writer
+    /// thread that owns the underlying writer (see `keeper_writer_loop`). Keeping
+    /// the writer off the request path means no handler holds a lock across a
+    /// blocking write.
+    write_tx: SyncSender<Vec<u8>>,
     killer: Mutex<Box<dyn ChildKiller + Send + Sync>>,
     pipe: Mutex<Option<KeeperPipe>>,
     sequence: AtomicU64,
@@ -672,9 +686,12 @@ impl KeeperSurface {
         }
     }
 
-    fn route_output(&mut self, bytes: &[u8]) -> ActivityState {
+    /// Ingest PTY output and return the resulting activity plus any terminal
+    /// replies the emulator generated (device-attribute / cursor-position / mode
+    /// queries). The caller writes the replies back to the PTY master.
+    fn route_output(&mut self, bytes: &[u8]) -> (ActivityState, Vec<u8>) {
         self.router.route_output(self.backend.as_mut(), bytes);
-        self.backend.take_activity()
+        self.backend.take_events()
     }
 
     fn resize(&mut self, size: PtySize) {
@@ -956,10 +973,20 @@ pub fn run_runtime_keeper(cli: RuntimeKeeperCli) -> Result<()> {
         .take_writer()
         .map_err(|error| MuxError::pty(error.to_string()))?;
 
+    // A single writer thread owns the writer and drains every PTY-master write
+    // (client input and emulator replies) from this bounded queue, so no request
+    // handler holds a lock across a blocking write. The only senders live in the
+    // shared runtime; the thread ends once the last runtime reference is dropped.
+    let (write_tx, write_rx) = sync_channel(KEEPER_WRITE_QUEUE_CAPACITY);
+    let writer_join = thread::Builder::new()
+        .name(format!("keeper-writer-{}", cli.socket_path.display()))
+        .spawn(move || keeper_writer_loop(writer, write_rx))
+        .map_err(|error| MuxError::internal(error.to_string()))?;
+
     let runtime = Arc::new(KeeperRuntime {
         surface: Mutex::new(KeeperSurface::new(cli.size)),
         master: Mutex::new(pair.master),
-        writer: Mutex::new(writer),
+        write_tx,
         killer: Mutex::new(killer),
         pipe: Mutex::new(None),
         sequence: AtomicU64::new(0),
@@ -986,6 +1013,10 @@ pub fn run_runtime_keeper(cli: RuntimeKeeperCli) -> Result<()> {
 
     let _ = reader_join.join();
     let _ = wait_join.join();
+    // Drop the last runtime reference so the write channel disconnects and the
+    // writer thread finishes draining before we join it.
+    drop(runtime);
+    let _ = writer_join.join();
     Ok(())
 }
 
@@ -1116,13 +1147,13 @@ impl KeeperRuntime {
 
     fn write(&self, bytes: Vec<u8>) -> Result<()> {
         self.ensure_running()?;
-        let mut writer = self
-            .writer
-            .lock()
-            .map_err(|_| MuxError::internal("runtime keeper writer lock poisoned"))?;
-        writer.write_all(&bytes)?;
-        writer.flush()?;
-        Ok(())
+        // Hand off to the writer thread. This is a bounded send: it blocks only
+        // when the queue is full (backpressure from a child that is not reading
+        // its input) and never holds the writer, so it cannot stall other writers
+        // or the read loop. A send error means the writer thread is gone.
+        self.write_tx
+            .send(bytes)
+            .map_err(|_| MuxError::internal("runtime keeper writer channel closed"))
     }
 
     fn resize(&self, size: PtySize) -> Result<()> {
@@ -1262,7 +1293,7 @@ fn keeper_read_loop(runtime: Arc<KeeperRuntime>, mut reader: Box<dyn Read + Send
                     Err(_) => break,
                 };
                 let bytes = &buffer[..read];
-                let activity = surface.route_output(bytes);
+                let (activity, replies) = surface.route_output(bytes);
                 if let Ok(mut pipe) = runtime.pipe.lock()
                     && let Some(pipe) = pipe.as_mut()
                 {
@@ -1272,13 +1303,45 @@ fn keeper_read_loop(runtime: Arc<KeeperRuntime>, mut reader: Box<dyn Read + Send
                 if let Ok(mut state) = runtime.activity.lock() {
                     *state = activity;
                 }
+                // Release the surface lock before handing off the replies.
+                drop(surface);
+                if !replies.is_empty() {
+                    // Never write to the master from this thread: a blocked write
+                    // would stop draining child output and could deadlock. Hand
+                    // the reply to the writer thread without blocking; if the queue
+                    // is full (writer stuck on a full input buffer) drop the reply
+                    // — continuing to drain lets the child progress and recover.
+                    match runtime.write_tx.try_send(replies) {
+                        Ok(()) => {}
+                        Err(TrySendError::Full(_)) => {
+                            debug!("terminal query reply queue full; dropping reply");
+                        }
+                        Err(TrySendError::Disconnected(_)) => break,
+                    }
+                }
             }
             Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
             Err(_) => break,
         }
     }
 }
 
+/// Own the PTY-master writer and serialize every write submitted through the
+/// runtime's write queue (client input and emulator replies).
+///
+/// Running on its own thread means a master write that blocks (child not draining
+/// its input) applies backpressure to the queue rather than stalling any request
+/// handler or the read loop. A write can still fail once the child has exited
+/// while the queue drains; log at debug and keep going. Ends when every sender —
+/// all held in the shared runtime — is dropped.
+fn keeper_writer_loop(mut writer: Box<dyn Write + Send>, write_rx: Receiver<Vec<u8>>) {
+    for bytes in write_rx {
+        if let Err(error) = writer.write_all(&bytes).and_then(|()| writer.flush()) {
+            debug!(%error, "failed to write to pty master");
+        }
+    }
+}
+
 fn keeper_wait_loop(runtime: Arc<KeeperRuntime>, mut child: Box<dyn Child + Send + Sync>) {
     let exit_code = child.wait().ok().and_then(exit_status_code);
     if let Ok(mut state) = runtime.exit_code.lock() {

crates/embers-server/src/config.rs

@@ -4,6 +4,19 @@ use std::path::PathBuf;
 
 pub const SOCKET_ENV_VAR: &str = "EMBERS_SOCKET";
 
+/// `TERM` advertised to buffer child processes. Embers ships no terminfo of its
+/// own; alacritty's emulation is closest to xterm and `xterm-256color` exists
+/// everywhere. A user-supplied `TERM` in the buffer spawn env still overrides
+/// this (hints are merged after the base env).
+pub const TERM_ENV_VAR: &str = "TERM";
+/// Default `TERM` value injected into buffer children.
+pub const DEFAULT_TERM: &str = "xterm-256color";
+/// `COLORTERM` advertised to buffer child processes. Truthful at the parser
+/// level: alacritty accepts 24-bit SGR sequences.
+pub const COLORTERM_ENV_VAR: &str = "COLORTERM";
+/// Default `COLORTERM` value injected into buffer children.
+pub const DEFAULT_COLORTERM: &str = "truecolor";
+
 /// Environment variable overriding [`ResourceLimits::max_sessions`].
 pub const MAX_SESSIONS_ENV_VAR: &str = "EMBERS_MAX_SESSIONS";
 /// Environment variable overriding [`ResourceLimits::max_buffers`].
@@ -86,6 +99,14 @@ impl ServerConfig {
             SOCKET_ENV_VAR.to_owned(),
             socket_path.as_os_str().to_owned(),
         );
+        // Base terminal env for buffer children. User env hints are merged after
+        // this base (see `Server::spawn_buffer_runtime`), so a caller-specified
+        // TERM/COLORTERM still wins.
+        buffer_env.insert(TERM_ENV_VAR.to_owned(), OsString::from(DEFAULT_TERM));
+        buffer_env.insert(
+            COLORTERM_ENV_VAR.to_owned(),
+            OsString::from(DEFAULT_COLORTERM),
+        );
         let workspace_path = socket_path.with_extension("workspace.json");
         let runtime_dir = socket_path.with_extension("runtimes");
         Self {

crates/embers-server/src/terminal_backend.rs

@@ -57,7 +57,13 @@ pub trait TerminalBackend: Send {
     fn capture_scrollback(&self) -> Vec<String>;
     fn capture_scrollback_slice(&self, start_line: u64, line_count: u32) -> BackendScrollbackSlice;
     fn metadata(&self) -> BackendMetadata;
-    fn take_activity(&mut self) -> ActivityState;
+    /// Drain, under a single lock acquisition, the pending activity state and any
+    /// terminal replies the emulator generated while ingesting bytes
+    /// (device-attribute, cursor-position, and mode queries). The replies are
+    /// written back to the PTY master; the returned `Vec` is empty when there is
+    /// nothing pending. Combining the two drains keeps the read hot path to one
+    /// lock acquisition against the client-thread readers of the same state.
+    fn take_events(&mut self) -> (ActivityState, Vec<u8>);
     fn take_damage(&mut self) -> BackendDamage;
 }
 
@@ -106,6 +112,7 @@ struct BackendEventProxy {
 struct BackendEventState {
     title: Option<String>,
     bell_pending: bool,
+    pty_write: Vec<u8>,
 }
 
 impl BackendEventProxy {
@@ -128,6 +135,7 @@ impl EventListener for BackendEventProxy {
             Event::Title(title) => state.title = Some(title),
             Event::ResetTitle => state.title = None,
             Event::Bell => state.bell_pending = true,
+            Event::PtyWrite(text) => state.pty_write.extend_from_slice(text.as_bytes()),
             _ => {}
         }
     }
@@ -433,16 +441,20 @@ impl TerminalBackend for AlacrittyTerminalBackend {
         }
     }
 
-    fn take_activity(&mut self) -> ActivityState {
+    fn take_events(&mut self) -> (ActivityState, Vec<u8>) {
+        // Recover from a poisoned lock rather than crashing: the event state is
+        // plain data, and dropping replies here would leave inner apps waiting on
+        // query timeouts.
         let mut state = self
             .events
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner());
-        if std::mem::take(&mut state.bell_pending) {
+        let activity = if std::mem::take(&mut state.bell_pending) {
             ActivityState::Bell
         } else {
             ActivityState::Activity
-        }
+        };
+        (activity, std::mem::take(&mut state.pty_write))
     }
 
     fn take_damage(&mut self) -> BackendDamage {
@@ -647,8 +659,8 @@ mod tests {
             BackendMetadata::default()
         }
 
-        fn take_activity(&mut self) -> ActivityState {
-            ActivityState::Activity
+        fn take_events(&mut self) -> (ActivityState, Vec<u8>) {
+            (ActivityState::Activity, Vec::new())
         }
 
         fn take_damage(&mut self) -> BackendDamage {
@@ -990,11 +1002,69 @@ mod tests {
 
         let metadata = backend.metadata();
         assert_eq!(metadata.title.as_deref(), Some("embers"));
-        assert_eq!(backend.take_activity(), ActivityState::Bell);
+        assert_eq!(backend.take_events().0, ActivityState::Bell);
 
         let metadata = backend.metadata();
         assert_eq!(metadata.title.as_deref(), Some("embers"));
-        assert_eq!(backend.take_activity(), ActivityState::Activity);
+        assert_eq!(backend.take_events().0, ActivityState::Activity);
+    }
+
+    #[test]
+    fn da1_query_produces_device_attributes_reply() {
+        let mut backend = backend(PtySize::new(10, 2));
+        let _ = backend.take_damage();
+
+        backend.ingest_bytes(b"\x1b[c");
+
+        let reply = backend.take_events().1;
+        let text = String::from_utf8(reply).expect("reply is utf8");
+        assert!(text.starts_with("\x1b[?"), "reply: {text:?}");
+        assert!(text.ends_with('c'), "reply: {text:?}");
+
+        // Drained: a second call returns nothing.
+        assert!(backend.take_events().1.is_empty());
+    }
+
+    #[test]
+    fn dsr_cursor_position_report_reports_row_and_column() {
+        let mut backend = backend(PtySize::new(10, 2));
+        let _ = backend.take_damage();
+
+        backend.ingest_bytes(b"ab\x1b[6n");
+
+        let reply = backend.take_events().1;
+        let text = String::from_utf8(reply).expect("reply is utf8");
+        // Cursor sits after "ab" on row 1: CPR is ESC [ <row> ; <col> R.
+        assert_eq!(text, "\x1b[1;3R", "reply: {text:?}");
+    }
+
+    #[test]
+    fn decrqm_reports_bracketed_paste_mode() {
+        let mut backend = backend(PtySize::new(10, 2));
+        let _ = backend.take_damage();
+
+        backend.ingest_bytes(b"\x1b[?2004h\x1b[?2004$p");
+
+        let reply = backend.take_events().1;
+        let text = String::from_utf8(reply).expect("reply is utf8");
+        assert!(text.starts_with("\x1b[?2004;"), "reply: {text:?}");
+        assert!(text.ends_with("$y"), "reply: {text:?}");
+    }
+
+    #[test]
+    fn pty_writes_accumulate_alongside_title_and_bell() {
+        let mut backend = backend(PtySize::new(10, 2));
+        let _ = backend.take_damage();
+
+        backend.ingest_bytes(b"\x1b]0;embers\x07\x1b[c\x07");
+
+        let metadata = backend.metadata();
+        assert_eq!(metadata.title.as_deref(), Some("embers"));
+
+        // A single drain returns both the bell activity and the accumulated reply.
+        let (activity, reply) = backend.take_events();
+        assert_eq!(activity, ActivityState::Bell);
+        assert!(!reply.is_empty(), "device-attributes reply should survive");
     }
 
     #[test]