Keep PTY runtimes alive across server restarts

crates/embers-cli/Cargo.toml

@@ -18,6 +18,7 @@ name = "embers-cli"
 path = "src/bin/embers-cli.rs"
 
 [dependencies]
+base64.workspace = true
 clap.workspace = true
 embers-client = { path = "../embers-client" }
 embers-core = { path = "../embers-core" }

crates/embers-cli/src/lib.rs

@@ -1,13 +1,19 @@
 mod interactive;
 
+use std::ffi::OsString;
 use std::fs::{self, OpenOptions};
 use std::io::Write;
 use std::num::NonZeroU64;
 #[cfg(unix)]
+use std::os::unix::ffi::OsStringExt;
+#[cfg(unix)]
 use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
+#[cfg(windows)]
+use std::os::windows::ffi::OsStringExt;
 use std::path::{Path, PathBuf};
 use std::process::{Command as ProcessCommand, Stdio};
 
+use base64::Engine as _;
 use clap::{Parser, Subcommand};
 use embers_core::{
     BufferId, FloatGeometry, FloatingId, MuxError, NodeId, Result, SessionId, SplitDirection,
@@ -71,6 +77,21 @@ pub enum Command {
     },
     #[command(name = "__serve", hide = true)]
     Serve,
+    #[command(name = "__runtime-keeper", hide = true)]
+    RuntimeKeeper {
+        #[arg(long = "keeper-socket")]
+        keeper_socket: PathBuf,
+        #[arg(long)]
+        cols: u16,
+        #[arg(long)]
+        rows: u16,
+        #[arg(long)]
+        cwd: Option<PathBuf>,
+        #[arg(long = "env", value_parser = parse_env_arg)]
+        env: Vec<(String, OsString)>,
+        #[arg(last = true)]
+        command: Vec<String>,
+    },
     Ping {
         #[arg(default_value = "phase0")]
         payload: String,
@@ -226,9 +247,9 @@ async fn execute(socket: &Path, command: Command) -> Result<String> {
     let mut connection = CliConnection::connect(socket).await?;
 
     match command {
-        Command::Attach { .. } | Command::Serve => Err(MuxError::internal(
-            "interactive commands must be dispatched through run()",
-        )),
+        Command::Attach { .. } | Command::Serve | Command::RuntimeKeeper { .. } => Err(
+            MuxError::internal("interactive commands must be dispatched through run()"),
+        ),
         Command::Ping { payload } => {
             let response = connection
                 .request(ClientMessage::Ping(PingRequest {
@@ -621,31 +642,56 @@ async fn execute(socket: &Path, command: Command) -> Result<String> {
 }
 
 pub async fn run(cli: Cli) -> Result<()> {
-    let socket = resolve_socket_path(cli.socket.as_deref());
-    validate_runtime_socket_parent(&socket)?;
+    let Cli {
+        socket,
+        config,
+        command,
+        ..
+    } = cli;
 
-    match cli.command {
-        None => {
-            ensure_server_process(&socket).await?;
-            interactive::run(socket, None, cli.config).await
-        }
-        Some(Command::Attach { target }) => {
-            if !server_is_available(&socket).await {
-                return Err(MuxError::not_found(format!(
-                    "no embers server is listening on {}",
-                    socket.display()
-                )));
-            }
-            interactive::run(socket, target, cli.config).await
-        }
-        Some(Command::Serve) => run_server(socket).await,
-        Some(command) => {
-            ensure_server_process(&socket).await?;
-            let output = execute(&socket, command).await?;
-            if !output.is_empty() {
-                println!("{output}");
+    match command {
+        Some(Command::RuntimeKeeper {
+            keeper_socket,
+            cols,
+            rows,
+            cwd,
+            env,
+            command,
+        }) => embers_server::run_runtime_keeper(embers_server::RuntimeKeeperCli {
+            socket_path: keeper_socket,
+            command,
+            cwd,
+            env: env.into_iter().collect(),
+            size: embers_core::PtySize::new(cols, rows),
+        }),
+        command => {
+            let socket = resolve_socket_path(socket.as_deref());
+            validate_runtime_socket_parent(&socket)?;
+
+            match command {
+                None => {
+                    ensure_server_process(&socket).await?;
+                    interactive::run(socket, None, config).await
+                }
+                Some(Command::Attach { target }) => {
+                    if !server_is_available(&socket).await {
+                        return Err(MuxError::not_found(format!(
+                            "no embers server is listening on {}",
+                            socket.display()
+                        )));
+                    }
+                    interactive::run(socket, target, config).await
+                }
+                Some(Command::Serve) => run_server(socket).await,
+                Some(command) => {
+                    ensure_server_process(&socket).await?;
+                    let output = execute(&socket, command).await?;
+                    if !output.is_empty() {
+                        println!("{output}");
+                    }
+                    Ok(())
+                }
             }
-            Ok(())
         }
     }
 }
@@ -676,6 +722,46 @@ fn default_runtime_dir() -> PathBuf {
     PathBuf::from("/tmp").join(format!("embers-{}", effective_uid()))
 }
 
+fn parse_env_arg(value: &str) -> std::result::Result<(String, OsString), String> {
+    let Some((key, env_value)) = value.split_once('=') else {
+        return Err("expected KEY=VALUE".to_owned());
+    };
+    if key.is_empty() {
+        return Err("environment key must not be empty".to_owned());
+    }
+    Ok((key.to_owned(), decode_runtime_keeper_env_value(env_value)?))
+}
+
+fn decode_runtime_keeper_env_value(value: &str) -> std::result::Result<OsString, String> {
+    let Some(encoded) = value.strip_prefix("base64:") else {
+        return Ok(OsString::from(value));
+    };
+    let decoded = base64::engine::general_purpose::STANDARD
+        .decode(encoded)
+        .map_err(|error| format!("invalid base64 environment value: {error}"))?;
+    #[cfg(unix)]
+    {
+        Ok(OsString::from_vec(decoded))
+    }
+    #[cfg(windows)]
+    {
+        if decoded.len() % 2 != 0 {
+            return Err("invalid UTF-16LE environment value: odd-length byte sequence".to_owned());
+        }
+        let wide = decoded
+            .chunks_exact(2)
+            .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
+            .collect::<Vec<_>>();
+        Ok(OsString::from_wide(&wide))
+    }
+    #[cfg(all(not(unix), not(windows)))]
+    {
+        String::from_utf8(decoded)
+            .map(OsString::from)
+            .map_err(|error| format!("invalid UTF-8 environment value: {error}"))
+    }
+}
+
 #[cfg(unix)]
 fn effective_uid() -> u32 {
     unsafe { libc::geteuid() }
@@ -1585,9 +1671,20 @@ fn default_title(command: &[String], fallback: &str) -> String {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(windows)]
+    use base64::Engine as _;
     use clap::Parser;
     use embers_core::NodeId;
     use embers_protocol::{TabRecord, TabsRecord};
+    #[cfg(windows)]
+    use std::ffi::OsString;
+    #[cfg(unix)]
+    use std::ffi::OsString;
+    #[cfg(unix)]
+    use std::os::unix::ffi::OsStringExt;
+    #[cfg(windows)]
+    use std::os::windows::ffi::OsStringExt;
+    use std::path::Path;
 
     use super::{Cli, resolve_window_index, split_scoped_required, split_scoped_target};
 
@@ -1614,6 +1711,59 @@ mod tests {
         }
     }
 
+    #[test]
+    fn runtime_keeper_uses_distinct_keeper_socket_flag() {
+        let cli = Cli::try_parse_from([
+            "embers",
+            "__runtime-keeper",
+            "--socket",
+            "/tmp/global.sock",
+            "--keeper-socket",
+            "/tmp/keeper.sock",
+            "--cols",
+            "80",
+            "--rows",
+            "24",
+            "--",
+            "/bin/sh",
+        ])
+        .expect("cli parses");
+
+        assert_eq!(cli.socket.as_deref(), Some(Path::new("/tmp/global.sock")));
+        match cli.command {
+            Some(super::Command::RuntimeKeeper {
+                keeper_socket,
+                cols,
+                rows,
+                command,
+                ..
+            }) => {
+                assert_eq!(keeper_socket, Path::new("/tmp/keeper.sock"));
+                assert_eq!((cols, rows), (80, 24));
+                assert_eq!(command, vec!["/bin/sh"]);
+            }
+            other => panic!("expected runtime keeper command, got {other:?}"),
+        }
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn runtime_keeper_env_values_decode_base64_losslessly() {
+        let (key, value) = super::parse_env_arg("KEY=base64:AP8=").expect("env parses");
+        assert_eq!(key, "KEY");
+        assert_eq!(value, OsString::from_vec(vec![0, 255]));
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn runtime_keeper_env_values_decode_utf16le_losslessly() {
+        let encoded = base64::engine::general_purpose::STANDARD.encode([0x00, 0xD8, 0x61, 0x00]);
+        let (key, value) =
+            super::parse_env_arg(&format!("KEY=base64:{encoded}")).expect("env parses");
+        assert_eq!(key, "KEY");
+        assert_eq!(value, OsString::from_wide(&[0xD800, 0x0061]));
+    }
+
     #[test]
     fn scoped_targets_split_session_prefix() {
         assert_eq!(

crates/embers-cli/tests/interactive.rs

@@ -3,7 +3,7 @@ use std::path::Path;
 use std::time::Duration;
 
 use embers_core::PtySize;
-use embers_test_support::{PtyHarness, TestServer, cargo_bin, cargo_bin_path};
+use embers_test_support::{PtyHarness, TestServer, acquire_test_lock, cargo_bin, cargo_bin_path};
 use tempfile::tempdir;
 
 use crate::support::{run_cli, stdout};
@@ -165,6 +165,7 @@ fn first_client_id_finds_attached_row() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn embers_without_subcommand_starts_server_and_client() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let tempdir = tempdir().expect("tempdir");
     let socket_path = tempdir.path().join("embers.sock");
     let socket_arg = socket_path.to_string_lossy().into_owned();
@@ -204,6 +205,7 @@ async fn embers_without_subcommand_starts_server_and_client() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn attach_subcommand_connects_to_running_server() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let server = TestServer::start().await.expect("start server");
     let binary = cargo_bin_path("embers");
     let binary_dir = binary.parent().expect("binary dir");
@@ -248,6 +250,7 @@ async fn attach_subcommand_connects_to_running_server() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn client_commands_can_switch_and_detach_a_live_attached_client() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let server = TestServer::start().await.expect("start server");
 
     run_cli(&server, ["new-session", "main"]);
@@ -302,6 +305,7 @@ async fn client_commands_can_switch_and_detach_a_live_attached_client() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn page_up_enters_local_scrollback_and_shows_indicator() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let tempdir = tempdir().expect("tempdir");
     let socket_path = tempdir.path().join("embers.sock");
     let socket_arg = socket_path.to_string_lossy().into_owned();
@@ -321,6 +325,7 @@ async fn page_up_enters_local_scrollback_and_shows_indicator() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn local_selection_yank_emits_osc52_clipboard_sequence() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let tempdir = tempdir().expect("tempdir");
     let socket_path = tempdir.path().join("embers.sock");
     let socket_arg = socket_path.to_string_lossy().into_owned();

crates/embers-cli/tests/panes.rs

@@ -2,13 +2,14 @@ use std::time::Duration;
 
 use embers_core::RequestId;
 use embers_protocol::{BufferRequest, ClientMessage, ServerResponse};
-use embers_test_support::{TestConnection, TestServer};
+use embers_test_support::{TestConnection, TestServer, acquire_test_lock};
 use tokio::time::sleep;
 
 use crate::support::{run_cli, session_snapshot_by_name, stdout};
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn pane_commands_round_trip_through_cli() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let server = TestServer::start().await.expect("start server");
 
     run_cli(&server, ["new-session", "alpha"]);
@@ -129,6 +130,7 @@ async fn pane_commands_round_trip_through_cli() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn detached_buffers_can_be_listed_and_attached_via_cli() {
+    let _guard = acquire_test_lock().await.expect("acquire test lock");
     let server = TestServer::start().await.expect("start server");
 
     run_cli(&server, ["new-session", "alpha"]);

crates/embers-core/src/metadata.rs

@@ -1,7 +1,9 @@
 use std::path::PathBuf;
 use std::time::SystemTime;
 
-#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
 pub struct Timestamp(pub SystemTime);
 
 impl Timestamp {
@@ -16,15 +18,15 @@ impl Default for Timestamp {
     }
 }
 
-#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
 pub enum ActivityState {
     #[default]
     Idle,
     Activity,
     Bell,
 }
 
-#[derive(Clone, Debug, PartialEq, Eq)]
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct EntityMetadata {
     pub title: Option<String>,
     pub cwd: Option<PathBuf>,