Build interactive native workspace vertical slice

crates/lumbridge-pty/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "lumbridge-pty"
+description = "Portable local PTY process boundary for Lumbridge"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+
+[dependencies]
+portable-pty = "0.9.0"
+thiserror = "2.0"
+
+[target.'cfg(unix)'.dependencies]
+nix = { version = "0.28", features = ["process", "signal"] }
+
+[lints]
+workspace = true

crates/lumbridge-pty/src/lib.rs

@@ -0,0 +1,716 @@
+//! Portable local PTY process ownership for the Lumbridge runtime.
+//!
+//! This crate moves unparsed bytes between a child process and its PTY. Terminal
+//! emulation, scrollback, persistence, remote sessions, and secret injection are
+//! intentionally separate concerns.
+
+#![forbid(unsafe_code)]
+
+use portable_pty::{Child, CommandBuilder, MasterPty, PtySize, native_pty_system};
+use std::ffi::OsString;
+use std::io::{self, Read, Write};
+use std::num::NonZeroUsize;
+use std::path::PathBuf;
+use std::sync::mpsc::{self, Receiver, RecvTimeoutError, SyncSender};
+use std::thread::{self, JoinHandle};
+use std::time::{Duration, Instant};
+use thiserror::Error;
+
+const OUTPUT_CHUNK_BYTES: usize = 8 * 1024;
+const DEFAULT_OUTPUT_QUEUE_CHUNKS: usize = 64;
+const TERMINATION_GRACE: Duration = Duration::from_millis(250);
+const TERMINATION_POLL: Duration = Duration::from_millis(10);
+
+/// A validated terminal cell and pixel size.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct TerminalSize {
+    rows: u16,
+    columns: u16,
+    pixel_width: u16,
+    pixel_height: u16,
+}
+
+impl TerminalSize {
+    /// Creates a cell-only size. Rows and columns must both be non-zero.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::InvalidSize`] when either cell dimension is zero.
+    pub fn new(rows: u16, columns: u16) -> Result<Self, PtyError> {
+        Self::with_pixels(rows, columns, 0, 0)
+    }
+
+    /// Creates a size that also reports the viewport's pixel dimensions.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::InvalidSize`] when either cell dimension is zero.
+    pub fn with_pixels(
+        rows: u16,
+        columns: u16,
+        pixel_width: u16,
+        pixel_height: u16,
+    ) -> Result<Self, PtyError> {
+        if rows == 0 || columns == 0 {
+            return Err(PtyError::InvalidSize);
+        }
+
+        Ok(Self {
+            rows,
+            columns,
+            pixel_width,
+            pixel_height,
+        })
+    }
+
+    #[must_use]
+    pub const fn rows(self) -> u16 {
+        self.rows
+    }
+
+    #[must_use]
+    pub const fn columns(self) -> u16 {
+        self.columns
+    }
+}
+
+impl Default for TerminalSize {
+    fn default() -> Self {
+        Self {
+            rows: 24,
+            columns: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        }
+    }
+}
+
+impl From<TerminalSize> for PtySize {
+    fn from(size: TerminalSize) -> Self {
+        Self {
+            rows: size.rows,
+            cols: size.columns,
+            pixel_width: size.pixel_width,
+            pixel_height: size.pixel_height,
+        }
+    }
+}
+
+/// Credential-free process launch configuration.
+///
+/// The configuration deliberately has no arbitrary environment-variable API.
+/// A child receives only a small fixed allowlist from the Lumbridge process,
+/// plus terminal capability variables. Future secret injection belongs behind
+/// the runtime's secret-store boundary and must not be represented here.
+pub struct CommandConfig {
+    program: OsString,
+    arguments: Vec<OsString>,
+    working_directory: Option<PathBuf>,
+}
+
+impl CommandConfig {
+    /// Creates a launch configuration for a non-empty executable name or path.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::EmptyProgram`] for an empty executable name.
+    pub fn new(program: impl Into<OsString>) -> Result<Self, PtyError> {
+        let program = program.into();
+        if program.is_empty() {
+            return Err(PtyError::EmptyProgram);
+        }
+
+        Ok(Self {
+            program,
+            arguments: Vec::new(),
+            working_directory: None,
+        })
+    }
+
+    #[must_use]
+    pub fn arg(mut self, argument: impl Into<OsString>) -> Self {
+        self.arguments.push(argument.into());
+        self
+    }
+
+    #[must_use]
+    pub fn args<I, S>(mut self, arguments: I) -> Self
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<OsString>,
+    {
+        self.arguments.extend(arguments.into_iter().map(Into::into));
+        self
+    }
+
+    #[must_use]
+    pub fn working_directory(mut self, path: impl Into<PathBuf>) -> Self {
+        self.working_directory = Some(path.into());
+        self
+    }
+
+    fn into_builder(self) -> CommandBuilder {
+        let mut command = CommandBuilder::new(self.program);
+        command.args(self.arguments);
+        command.env_clear();
+
+        for name in inherited_environment_allowlist() {
+            if let Some(value) = std::env::var_os(name) {
+                command.env(name, value);
+            }
+        }
+
+        command.env("TERM", "xterm-256color");
+        command.env("COLORTERM", "truecolor");
+        if let Some(directory) = self.working_directory {
+            command.cwd(directory);
+        }
+        command
+    }
+}
+
+fn inherited_environment_allowlist() -> &'static [&'static str] {
+    &[
+        "HOME",
+        "LANG",
+        "LC_ALL",
+        "LC_CTYPE",
+        "LOGNAME",
+        "PATH",
+        "SHELL",
+        "TMPDIR",
+        "USER",
+        "XDG_CONFIG_HOME",
+        "XDG_DATA_HOME",
+        "XDG_RUNTIME_DIR",
+    ]
+}
+
+/// Spawn-time settings for a local PTY.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct PtyOptions {
+    pub size: TerminalSize,
+    /// Capacity in fixed-size output chunks. The reader blocks when full,
+    /// propagating backpressure to the kernel PTY rather than growing memory.
+    pub output_queue_chunks: NonZeroUsize,
+}
+
+impl PtyOptions {
+    #[must_use]
+    pub const fn new(size: TerminalSize) -> Self {
+        let output_queue_chunks = match NonZeroUsize::new(DEFAULT_OUTPUT_QUEUE_CHUNKS) {
+            Some(capacity) => capacity,
+            None => NonZeroUsize::MIN,
+        };
+        Self {
+            size,
+            output_queue_chunks,
+        }
+    }
+
+    #[must_use]
+    pub const fn with_output_queue_chunks(mut self, capacity: NonZeroUsize) -> Self {
+        self.output_queue_chunks = capacity;
+        self
+    }
+}
+
+impl Default for PtyOptions {
+    fn default() -> Self {
+        Self::new(TerminalSize::default())
+    }
+}
+
+/// A single item delivered by the bounded PTY output queue.
+#[derive(Debug, Eq, PartialEq)]
+pub enum OutputEvent {
+    /// Unparsed bytes read from the PTY. A chunk is never larger than 8 KiB.
+    Data(Vec<u8>),
+    /// The PTY slave closed normally.
+    Eof,
+    /// The reader stopped because the operating-system read failed.
+    ReadFailed(PtyReadError),
+}
+
+/// A redacted, cloneable description of a PTY reader failure.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct PtyReadError {
+    pub kind: io::ErrorKind,
+    pub message: String,
+}
+
+/// A completed child-process status.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ExitStatus {
+    pub code: u32,
+    pub signal: Option<String>,
+}
+
+impl ExitStatus {
+    #[must_use]
+    pub const fn success(&self) -> bool {
+        self.code == 0 && self.signal.is_none()
+    }
+}
+
+impl From<portable_pty::ExitStatus> for ExitStatus {
+    fn from(status: portable_pty::ExitStatus) -> Self {
+        Self {
+            code: status.exit_code(),
+            signal: status.signal().map(ToOwned::to_owned),
+        }
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum PtyError {
+    #[error("PTY program cannot be empty")]
+    EmptyProgram,
+    #[error("PTY rows and columns must both be non-zero")]
+    InvalidSize,
+    #[error("failed to open the local PTY: {0}")]
+    Open(String),
+    #[error("failed to clone the local PTY reader: {0}")]
+    CloneReader(String),
+    #[error("failed to take the local PTY writer: {0}")]
+    TakeWriter(String),
+    #[error("failed to spawn the local PTY child: {0}")]
+    Spawn(String),
+    #[error("the PTY session is closed")]
+    Closed,
+    #[error("PTY I/O failed")]
+    Io(#[source] io::Error),
+    #[error("PTY child did not exit within {0:?}")]
+    ExitTimeout(Duration),
+    #[error("PTY output was not available before the timeout")]
+    OutputTimeout,
+    #[error("PTY output reader disconnected without an EOF event")]
+    OutputDisconnected,
+}
+
+/// Owns one local PTY, its child process, and its bounded output queue.
+///
+/// A session is intentionally single-owner. The future runtime actor performs
+/// input ordering and forwards output events to terminal emulation.
+pub struct PtySession {
+    master: Option<Box<dyn MasterPty + Send>>,
+    writer: Option<Box<dyn Write + Send>>,
+    child: Box<dyn Child + Send + Sync>,
+    process_id: Option<u32>,
+    output: Option<Receiver<OutputEvent>>,
+    reader_thread: Option<JoinHandle<()>>,
+    exit_status: Option<ExitStatus>,
+}
+
+impl PtySession {
+    /// Opens a native PTY and spawns the configured child inside it.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error when the PTY, reader, writer, child, or reader thread
+    /// cannot be created.
+    pub fn spawn(config: CommandConfig, options: PtyOptions) -> Result<Self, PtyError> {
+        let pty_system = native_pty_system();
+        let pair = pty_system
+            .openpty(options.size.into())
+            .map_err(|error| PtyError::Open(error.to_string()))?;
+        let reader = pair
+            .master
+            .try_clone_reader()
+            .map_err(|error| PtyError::CloneReader(error.to_string()))?;
+        let writer = pair
+            .master
+            .take_writer()
+            .map_err(|error| PtyError::TakeWriter(error.to_string()))?;
+        let (sender, output) = mpsc::sync_channel(options.output_queue_chunks.get());
+        let reader_thread = thread::Builder::new()
+            .name("lumbridge-pty-reader".to_owned())
+            .spawn(move || read_output(reader, &sender))
+            .map_err(PtyError::Io)?;
+        let child = pair
+            .slave
+            .spawn_command(config.into_builder())
+            .map_err(|error| PtyError::Spawn(error.to_string()))?;
+        let process_id = child.process_id();
+        drop(pair.slave);
+
+        Ok(Self {
+            master: Some(pair.master),
+            writer: Some(writer),
+            child,
+            process_id,
+            output: Some(output),
+            reader_thread: Some(reader_thread),
+            exit_status: None,
+        })
+    }
+
+    #[must_use]
+    pub const fn process_id(&self) -> Option<u32> {
+        self.process_id
+    }
+
+    /// Writes ordered raw input bytes to the PTY and flushes the writer.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::Closed`] after input has closed, or [`PtyError::Io`]
+    /// when the operating-system write fails.
+    pub fn write_all(&mut self, bytes: &[u8]) -> Result<(), PtyError> {
+        let writer = self.writer.as_mut().ok_or(PtyError::Closed)?;
+        writer.write_all(bytes).map_err(PtyError::Io)?;
+        writer.flush().map_err(PtyError::Io)
+    }
+
+    /// Applies a validated terminal size to the PTY.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::Closed`] after session cleanup, or [`PtyError::Io`]
+    /// when the platform resize operation fails.
+    pub fn resize(&self, size: TerminalSize) -> Result<(), PtyError> {
+        self.master
+            .as_ref()
+            .ok_or(PtyError::Closed)?
+            .resize(size.into())
+            .map_err(|error| PtyError::Io(io::Error::other(error.to_string())))
+    }
+
+    /// Receives the next bounded output event before `timeout` elapses.
+    ///
+    /// # Errors
+    ///
+    /// Returns a closed, timeout, or unexpected-disconnection error when no
+    /// event can be delivered.
+    pub fn recv_output_timeout(&self, timeout: Duration) -> Result<OutputEvent, PtyError> {
+        let output = self.output.as_ref().ok_or(PtyError::Closed)?;
+        match output.recv_timeout(timeout) {
+            Ok(event) => Ok(event),
+            Err(RecvTimeoutError::Timeout) => Err(PtyError::OutputTimeout),
+            Err(RecvTimeoutError::Disconnected) => Err(PtyError::OutputDisconnected),
+        }
+    }
+
+    /// Polls the child without blocking and caches a completed status.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::Io`] when the platform child poll fails.
+    pub fn try_wait(&mut self) -> Result<Option<ExitStatus>, PtyError> {
+        if let Some(status) = &self.exit_status {
+            return Ok(Some(status.clone()));
+        }
+
+        let status: Option<ExitStatus> =
+            self.child.try_wait().map_err(PtyError::Io)?.map(Into::into);
+        if let Some(status) = &status {
+            self.exit_status = Some(status.clone());
+        }
+        Ok(status)
+    }
+
+    /// Blocks until the session leader exits and caches its status.
+    ///
+    /// Callers should continue consuming output from another runtime task when
+    /// a child may produce more than the bounded queue can hold.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::Io`] when the platform wait fails.
+    pub fn wait(&mut self) -> Result<ExitStatus, PtyError> {
+        if let Some(status) = &self.exit_status {
+            return Ok(status.clone());
+        }
+
+        let status = ExitStatus::from(self.child.wait().map_err(PtyError::Io)?);
+        self.exit_status = Some(status.clone());
+        Ok(status)
+    }
+
+    /// Polls until the child exits or the supplied timeout expires.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::ExitTimeout`] on expiry or [`PtyError::Io`] when a
+    /// platform child poll fails.
+    pub fn wait_timeout(&mut self, timeout: Duration) -> Result<ExitStatus, PtyError> {
+        let deadline = Instant::now() + timeout;
+        loop {
+            if let Some(status) = self.try_wait()? {
+                return Ok(status);
+            }
+            if Instant::now() >= deadline {
+                return Err(PtyError::ExitTimeout(timeout));
+            }
+            thread::sleep(TERMINATION_POLL.min(deadline.saturating_duration_since(Instant::now())));
+        }
+    }
+
+    /// Terminates the PTY's process tree and waits for the session leader.
+    ///
+    /// Unix children run in the session/process group created by
+    /// `portable-pty`; the whole group is signalled so a shell's foreground
+    /// process cannot be orphaned. Other platforms use the portable child
+    /// killer.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`PtyError::Io`] when signalling or waiting for the child fails.
+    pub fn terminate(&mut self) -> Result<ExitStatus, PtyError> {
+        self.close_input();
+        terminate_process_tree(self.process_id, self.child.as_mut())?;
+        self.wait()
+    }
+
+    /// Closes the input half without terminating the child.
+    pub fn close_input(&mut self) {
+        self.writer.take();
+    }
+
+    fn disconnect_output(&mut self) {
+        self.output.take();
+    }
+
+    fn join_finished_reader(&mut self) {
+        if self
+            .reader_thread
+            .as_ref()
+            .is_some_and(JoinHandle::is_finished)
+            && let Some(reader_thread) = self.reader_thread.take()
+        {
+            let _ = reader_thread.join();
+        }
+    }
+}
+
+impl Drop for PtySession {
+    fn drop(&mut self) {
+        // Disconnect first so a reader blocked by queue backpressure can exit.
+        self.disconnect_output();
+        self.close_input();
+        if terminate_process_tree(self.process_id, self.child.as_mut()).is_ok() {
+            let _ = self.child.wait();
+        } else {
+            // Destructors must not hang indefinitely when platform signalling
+            // itself failed. The child handle is still dropped below.
+            let _ = self.child.try_wait();
+        }
+        self.master.take();
+        self.join_finished_reader();
+    }
+}
+
+fn read_output(mut reader: Box<dyn Read + Send>, sender: &SyncSender<OutputEvent>) {
+    let mut buffer = vec![0_u8; OUTPUT_CHUNK_BYTES];
+    loop {
+        match reader.read(&mut buffer) {
+            Ok(0) => {
+                let _ = sender.send(OutputEvent::Eof);
+                return;
+            }
+            Ok(length) => {
+                if sender
+                    .send(OutputEvent::Data(buffer[..length].to_vec()))
+                    .is_err()
+                {
+                    return;
+                }
+            }
+            Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
+            Err(error) => {
+                let _ = sender.send(OutputEvent::ReadFailed(PtyReadError {
+                    kind: error.kind(),
+                    message: error.to_string(),
+                }));
+                return;
+            }
+        }
+    }
+}
+
+#[cfg(unix)]
+fn terminate_process_tree(
+    process_id: Option<u32>,
+    child: &mut (dyn Child + Send + Sync),
+) -> Result<(), PtyError> {
+    use nix::errno::Errno;
+    use nix::sys::signal::{Signal, killpg};
+    use nix::unistd::Pid;
+
+    let Some(process_id) = process_id.and_then(|id| i32::try_from(id).ok()) else {
+        child.kill().map_err(PtyError::Io)?;
+        return Ok(());
+    };
+    let group = Pid::from_raw(process_id);
+
+    if let Err(error) = killpg(group, Signal::SIGHUP)
+        && error != Errno::ESRCH
+    {
+        return Err(PtyError::Io(io::Error::from_raw_os_error(error as i32)));
+    }
+
+    let deadline = Instant::now() + TERMINATION_GRACE;
+    while Instant::now() < deadline {
+        if child.try_wait().map_err(PtyError::Io)?.is_some() {
+            break;
+        }
+        thread::sleep(TERMINATION_POLL);
+    }
+
+    // Always address the group. The leader may have exited while a foreground
+    // descendant that ignored SIGHUP remains attached to the terminal.
+    if let Err(error) = killpg(group, Signal::SIGKILL)
+        && error != Errno::ESRCH
+    {
+        return Err(PtyError::Io(io::Error::from_raw_os_error(error as i32)));
+    }
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn terminate_process_tree(
+    _process_id: Option<u32>,
+    child: &mut (dyn Child + Send + Sync),
+) -> Result<(), PtyError> {
+    child.kill().map_err(PtyError::Io)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        CommandConfig, ExitStatus, OutputEvent, PtyError, PtyOptions, PtySession, TerminalSize,
+    };
+    use std::num::NonZeroUsize;
+    use std::time::{Duration, Instant};
+
+    const TEST_TIMEOUT: Duration = Duration::from_secs(3);
+
+    fn shell(script: &str) -> CommandConfig {
+        CommandConfig::new("/bin/sh")
+            .expect("/bin/sh is a valid program")
+            .args(["-c", script])
+    }
+
+    fn collect_until_eof(session: &PtySession) -> Vec<u8> {
+        let deadline = Instant::now() + TEST_TIMEOUT;
+        let mut output = Vec::new();
+        loop {
+            let remaining = deadline.saturating_duration_since(Instant::now());
+            assert!(!remaining.is_zero(), "timed out collecting PTY output");
+            match session.recv_output_timeout(remaining) {
+                Ok(OutputEvent::Data(bytes)) => {
+                    assert!(bytes.len() <= super::OUTPUT_CHUNK_BYTES);
+                    output.extend(bytes);
+                }
+                Ok(OutputEvent::Eof) => return output,
+                Ok(OutputEvent::ReadFailed(error)) => panic!("PTY read failed: {error:?}"),
+                Err(error) => panic!("PTY output failed: {error}"),
+            }
+        }
+    }
+
+    #[test]
+    fn captures_raw_output_and_nonzero_exit() {
+        let mut session = PtySession::spawn(
+            shell("printf 'lumbridge-ready\\n'; exit 7"),
+            PtyOptions::default(),
+        )
+        .expect("spawn synthetic shell");
+
+        let output = collect_until_eof(&session);
+        let status = session.wait_timeout(TEST_TIMEOUT).expect("shell exits");
+
+        assert!(String::from_utf8_lossy(&output).contains("lumbridge-ready"));
+        assert_eq!(
+            status,
+            ExitStatus {
+                code: 7,
+                signal: None
+            }
+        );
+    }
+
+    #[test]
+    fn writes_input_and_reports_resize_to_child() {
+        let initial = TerminalSize::new(24, 80).expect("valid initial size");
+        let resized = TerminalSize::new(41, 101).expect("valid resized size");
+        let mut session = PtySession::spawn(
+            shell("IFS= read -r line; stty size; printf 'got:%s\\n' \"$line\""),
+            PtyOptions::new(initial),
+        )
+        .expect("spawn synthetic shell");
+
+        session.resize(resized).expect("resize PTY");
+        session
+            .write_all(b"hello-lumbridge\n")
+            .expect("write PTY input");
+
+        let output = String::from_utf8_lossy(&collect_until_eof(&session)).replace('\r', "");
+        let status = session.wait_timeout(TEST_TIMEOUT).expect("shell exits");
+        assert!(output.contains("41 101"), "unexpected output: {output:?}");
+        assert!(
+            output.contains("got:hello-lumbridge"),
+            "unexpected output: {output:?}"
+        );
+        assert!(status.success());
+    }
+
+    #[test]
+    fn bounded_queue_preserves_large_output_when_consumed() {
+        let options = PtyOptions::default()
+            .with_output_queue_chunks(NonZeroUsize::new(1).expect("non-zero capacity"));
+        let mut session = PtySession::spawn(
+            shell("i=0; while [ $i -lt 2048 ]; do printf '0123456789abcdef'; i=$((i+1)); done"),
+            options,
+        )
+        .expect("spawn synthetic producer");
+
+        let output = collect_until_eof(&session);
+        let status = session.wait_timeout(TEST_TIMEOUT).expect("producer exits");
+        assert_eq!(output.len(), 32 * 1024);
+        assert!(status.success());
+    }
+
+    #[test]
+    fn terminate_stops_a_hung_child_with_a_timeout_bound() {
+        let mut session =
+            PtySession::spawn(shell("trap '' HUP; exec sleep 30"), PtyOptions::default())
+                .expect("spawn synthetic sleeper");
+
+        let started = Instant::now();
+        let status = session.terminate().expect("terminate process tree");
+        assert!(!status.success());
+        assert!(started.elapsed() < TEST_TIMEOUT);
+    }
+
+    #[test]
+    fn rejects_invalid_dimensions_and_empty_program() {
+        assert!(matches!(
+            TerminalSize::new(0, 80),
+            Err(PtyError::InvalidSize)
+        ));
+        assert!(matches!(
+            CommandConfig::new(""),
+            Err(PtyError::EmptyProgram)
+        ));
+    }
+
+    #[test]
+    fn child_environment_is_minimal_and_excludes_provider_credentials() {
+        let command = shell("exit 0").into_builder();
+
+        assert_eq!(
+            command.get_env("TERM"),
+            Some(std::ffi::OsStr::new("xterm-256color"))
+        );
+        for forbidden in [
+            "ANTHROPIC_API_KEY",
+            "CEREBRAS_API_KEY",
+            "GEMINI_API_KEY",
+            "GROQ_API_KEY",
+            "OPENAI_API_KEY",
+        ] {
+            assert_eq!(command.get_env(forbidden), None, "inherited {forbidden}");
+        }
+    }
+}