Add buffer and node cli actions

Cargo.toml

@@ -22,6 +22,7 @@ base64 = "0.22"
 clap = { version = "4.5", features = ["derive"] }
 criterion = "0.7"
 directories = "6"
+fastrand = "2"
 flatbuffers = "=25.12.19"
 libc = "0.2"
 portable-pty = "0.9"
@@ -31,6 +32,7 @@ rhai = "1"
 rhai-autodocs = "0.11"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+shell-words = "1"
 tempfile = "3"
 thiserror = "2"
 tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] }

crates/embers-cli/Cargo.toml

@@ -25,13 +25,15 @@ embers-core = { path = "../embers-core" }
 embers-protocol = { path = "../embers-protocol" }
 embers-server = { path = "../embers-server" }
 libc.workspace = true
+serde_json.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 unicode-width.workspace = true
 
 [dev-dependencies]
 assert_cmd.workspace = true
 embers-test-support = { path = "../embers-test-support" }
+filetime = "0.2"
 predicates.workspace = true
 tempfile.workspace = true
 

crates/embers-cli/src/interactive.rs

@@ -1,3 +1,4 @@
+use std::fs;
 use std::io::{self, Write};
 use std::os::fd::AsRawFd;
 use std::path::{Path, PathBuf};
@@ -17,6 +18,7 @@ const DEFAULT_SESSION_NAME: &str = "main";
 const KEY_SEQUENCE_TIMEOUT: Duration = Duration::from_millis(15);
 const KEY_SEQUENCE_CONTINUATION_TIMEOUT: Duration = Duration::from_millis(2);
 const EVENT_POLL_INTERVAL: Duration = Duration::from_millis(20);
+const CONFIG_WATCH_POLL_INTERVAL: Duration = Duration::from_millis(250);
 const BRACKETED_PASTE_END: &[u8] = b"\x1b[201~";
 const TERMINAL_ENTER_BASE_SEQUENCE: &str =
     "\x1b[?1049h\x1b[?1004h\x1b[?2004h\x1b[?25l\x1b[2J\x1b[H";
@@ -37,11 +39,13 @@ pub async fn run(
     let mut session_id = Some(initial_session_id);
     let config = ConfigManager::from_process(config_path)
         .map_err(|error| MuxError::invalid_input(error.to_string()))?;
+    let watched_config_path = config.active_source().path.clone();
     let mut configured = ConfiguredClient::new(client, config);
 
     let mut terminal = TerminalGuard::enter(mouse_capture_enabled(&configured))?;
     let (input_tx, mut input_rx) = mpsc::unbounded_channel();
-    let _input_thread = spawn_input_thread(input_tx)?;
+    let _input_thread = spawn_input_thread(input_tx.clone())?;
+    let _config_thread = spawn_config_thread(watched_config_path, input_tx)?;
 
     let mut terminal_size = terminal.size()?;
     let mut dirty = true;
@@ -117,6 +121,16 @@ pub async fn run(
                         dirty = true;
                     }
                 }
+                Ok(TerminalEvent::ConfigChanged) => match configured.reload_config_if_changed() {
+                    Ok(true) => {
+                        terminal.write_bytes(&drain_terminal_output(&mut configured))?;
+                        dirty = true;
+                    }
+                    Ok(false) => {}
+                    Err(_) => {
+                        dirty = true;
+                    }
+                },
                 Ok(TerminalEvent::InputClosed) => return Ok(()),
                 Ok(TerminalEvent::InputError(message)) => {
                     return Err(MuxError::transport(message));
@@ -133,9 +147,11 @@ pub async fn run(
             continue;
         }
 
-        match tokio::time::timeout(EVENT_POLL_INTERVAL, configured.process_next_event()).await {
-            Ok(result) => {
-                let event = result?;
+        match configured
+            .process_next_event_timeout(EVENT_POLL_INTERVAL)
+            .await?
+        {
+            Some(event) => {
                 match switched_session_id(&event, attached_client_id) {
                     SwitchedSession::Switched(next_session_id) => {
                         ensure_root_window(configured.client_mut(), next_session_id).await?;
@@ -149,7 +165,7 @@ pub async fn run(
                 terminal.write_bytes(&drain_terminal_output(&mut configured))?;
                 dirty = true;
             }
-            Err(_) => {
+            None => {
                 continue;
             }
         }
@@ -388,6 +404,7 @@ enum TerminalEvent {
     Mouse(MouseEvent),
     Paste(Vec<u8>),
     Focus(bool),
+    ConfigChanged,
     InputClosed,
     InputError(String),
 }
@@ -424,6 +441,59 @@ fn spawn_input_thread(
         .map_err(|error| MuxError::internal(format!("failed to spawn input thread: {error}")))
 }
 
+fn spawn_config_thread(
+    config_path: Option<PathBuf>,
+    tx: mpsc::UnboundedSender<TerminalEvent>,
+) -> Result<Option<std::thread::JoinHandle<()>>> {
+    let Some(config_path) = config_path else {
+        return Ok(None);
+    };
+
+    let handle = thread::Builder::new()
+        .name("embers-config".to_owned())
+        .spawn(move || {
+            let mut last_modified = config_modified(&config_path);
+            let mut pending_change = None;
+            let mut missing_polls = 0usize;
+            loop {
+                thread::sleep(CONFIG_WATCH_POLL_INTERVAL);
+                let observed_modified = config_modified(&config_path);
+                if observed_modified.is_none() && last_modified.is_some() {
+                    missing_polls = missing_polls.saturating_add(1);
+                    if missing_polls < 2 {
+                        continue;
+                    }
+                } else {
+                    missing_polls = 0;
+                }
+
+                if observed_modified == last_modified {
+                    pending_change = None;
+                    continue;
+                }
+
+                if pending_change == Some(observed_modified) {
+                    last_modified = observed_modified;
+                    pending_change = None;
+                    if tx.send(TerminalEvent::ConfigChanged).is_err() {
+                        break;
+                    }
+                } else {
+                    pending_change = Some(observed_modified);
+                }
+            }
+        })
+        .map_err(|error| MuxError::internal(format!("failed to spawn config thread: {error}")))?;
+
+    Ok(Some(handle))
+}
+
+fn config_modified(path: &Path) -> Option<std::time::SystemTime> {
+    fs::metadata(path)
+        .and_then(|metadata| metadata.modified())
+        .ok()
+}
+
 fn read_terminal_event(fd: libc::c_int) -> Result<Option<TerminalEvent>> {
     let Some(first) = read_byte(fd)? else {
         return Ok(None);
@@ -576,10 +646,7 @@ fn mouse_button(code: u16) -> Option<MouseButton> {
 
 fn read_bracketed_paste(fd: libc::c_int) -> Result<Vec<u8>> {
     let mut bytes = Vec::new();
-    loop {
-        let Some(next) = read_byte(fd)? else {
-            break;
-        };
+    while let Some(next) = read_byte(fd)? {
         bytes.push(next);
         if bytes.ends_with(BRACKETED_PASTE_END) {
             let new_len = bytes.len() - BRACKETED_PASTE_END.len();