feat(env): run programs in pseudo-terminals for interactive terminals

The daemon gets pty, ptyWrite and ptyResize: raw terminal output as events, input in arrival order, SIGWINCH on resize, and the process group killed on cancel or a lost connection. openTerminal() is the client, localDaemon() the daemon for this machine. Unix only; Windows answers ENOTSUP.

packages/env/README.md

@@ -33,6 +33,11 @@ checking against the application's own known-hosts file under a fixed alias, no
 connections or configured commands, and without forwarding the local locale. On Windows, detection and deployment go
 through PowerShell, and the daemon starts through the server's default shell (cmd.exe or PowerShell).
 
+`openTerminal(connection, options)` runs a program, by default the user's login shell, in a pseudo-terminal of the
+daemon: raw output, input, resizing and its exit code, for interactive terminals. `localDaemon()` is the daemon for this
+machine, for terminals without SSH (`new Connection({ command: [localDaemon()] })`). Terminals are not supported on
+Windows yet.
+
 Supported remote systems: Linux, macOS, Android (Termux) and Windows, on x86-64 and arm64. On Windows, string commands
 run through Git Bash as `NodeExecutionEnv` runs them there; argv commands run directly.
 

packages/env/daemon/src/exec.rs

@@ -57,7 +57,7 @@ impl Control {
         }
     }
 
-    fn attach(&self, sender: Sender<Message>) {
+    pub fn attach(&self, sender: Sender<Message>) {
         *self.sender.lock().unwrap() = Some(sender);
     }
 }

packages/env/daemon/src/main.rs

@@ -7,6 +7,7 @@ mod exec;
 mod frame;
 mod fs;
 mod output;
+mod pty;
 mod scan;
 mod sys;
 mod watch;
@@ -74,6 +75,7 @@ struct Server {
     next_handle: AtomicU64,
     controls: Mutex<HashMap<u32, Arc<Control>>>,
     groups: exec::Groups,
+    terminals: pty::Terminals,
 }
 
 fn now_ms() -> u64 {
@@ -253,6 +255,14 @@ impl Server {
                 &self.groups,
             )),
             "watch" => plain(watch::run(id, json, &self.output, control)),
+            "pty" => plain(pty::run(
+                id,
+                pty::PtyRequest::from_json(json)?,
+                &self.output,
+                control,
+                &self.groups,
+                &self.terminals,
+            )),
             _ => Err(Failure::new("EINVAL", format!("unknown operation {op}"))),
         }
     }
@@ -345,7 +355,7 @@ fn start_workers() -> Sender<Job> {
 }
 
 fn respond(server: &Server, id: u32, request: Frame, control: Arc<Control>) {
-    let is_exec = request.json["op"].as_str() == Some("exec");
+    let is_exec = matches!(request.json["op"].as_str(), Some("exec" | "pty"));
     let reply = match server.dispatch(id, &request.json, request.payload, &control) {
         Ok((json, payload)) => Frame::with_payload(RESULT, id, json, payload),
         Err(failure) => Frame::new(ERROR, id, failure.to_json()),
@@ -390,13 +400,25 @@ fn serve_frames(server: &Arc<Server>, input: &mut impl Read) -> io::Result<()> {
     let jobs = start_workers();
     while let Some(request) = frame::read_frame(input)? {
         match request.kind {
+            REQUEST if matches!(request.json["op"].as_str(), Some("ptyWrite" | "ptyResize")) => {
+                // Answered here, in arrival order: through the worker pool, two writes could swap.
+                let reply = if request.json["op"].as_str() == Some("ptyWrite") {
+                    pty::write(&server.terminals, &request.json, request.payload)
+                } else {
+                    pty::resize(&server.terminals, &request.json)
+                };
+                server.output.control(match reply {
+                    Ok(json) => Frame::new(RESULT, request.id, json),
+                    Err(failure) => Frame::new(ERROR, request.id, failure.to_json()),
+                });
+            }
             REQUEST => {
                 let id = request.id;
                 // Registered before the request runs, so a cancel right behind it is not lost.
                 let control = Arc::new(Control::default());
                 server.controls.lock().unwrap().insert(id, control.clone());
                 // Commands and watchers last; they get their own threads (watchers recurse through trees).
-                if matches!(request.json["op"].as_str(), Some("exec" | "watch")) {
+                if matches!(request.json["op"].as_str(), Some("exec" | "watch" | "pty")) {
                     let worker = server.clone();
                     let spawned = thread::Builder::new()
                         .stack_size(16 * 1024 * 1024)
@@ -455,6 +477,7 @@ fn serve(token: &str) -> io::Result<()> {
         next_handle: AtomicU64::new(1),
         controls: Mutex::new(HashMap::new()),
         groups: Arc::new(Mutex::new(Default::default())),
+        terminals: Arc::new(Mutex::new(HashMap::new())),
     });
     let last_seen = Arc::new(AtomicU64::new(now_ms()));
     {

packages/env/daemon/src/pty.rs

@@ -0,0 +1,498 @@
+//! `pty`: a program in a pseudo-terminal, for interactive terminals. Unlike `exec`, output is the terminal's raw
+//! bytes, and the program reads input: `ptyWrite` and `ptyResize` address the running `pty` request by its id.
+
+use crate::errors::Failure;
+use crate::exec::{Control, Groups};
+#[cfg(unix)]
+use crate::exec::{Message, kill_group};
+#[cfg(unix)]
+use crate::frame::{EVENT, Frame};
+use crate::output::Output;
+use serde_json::{Map, Value, json};
+use std::collections::HashMap;
+use std::sync::mpsc::Sender;
+use std::sync::{Arc, Mutex};
+
+/// After the program exits, output still arriving is collected until this much silence.
+#[cfg(unix)]
+const EXIT_GRACE: std::time::Duration = std::time::Duration::from_millis(100);
+/// Largest terminal size accepted, in cells.
+const MAX_CELLS: u64 = 4096;
+
+/// What `ptyWrite` and `ptyResize` reach: the terminal of one running `pty` request.
+pub struct Terminal {
+    input: Sender<Vec<u8>>,
+    #[cfg(unix)]
+    master: Arc<std::fs::File>,
+}
+
+/// Running terminals by the id of their `pty` request.
+pub type Terminals = Arc<Mutex<HashMap<u32, Arc<Terminal>>>>;
+
+#[cfg_attr(not(unix), allow(dead_code))]
+pub struct PtyRequest {
+    argv: Option<Vec<String>>,
+    cwd: String,
+    env: Map<String, Value>,
+    inherit_env: bool,
+    cols: u16,
+    rows: u16,
+}
+
+fn size(json: &Value, key: &str) -> Result<u16, Failure> {
+    match json[key].as_u64() {
+        Some(value) if (1..=MAX_CELLS).contains(&value) => Ok(value as u16),
+        _ => Err(Failure::new(
+            "EINVAL",
+            format!("{key} must be between 1 and {MAX_CELLS}"),
+        )),
+    }
+}
+
+impl PtyRequest {
+    pub fn from_json(json: &Value) -> Result<PtyRequest, Failure> {
+        let argv = match &json["argv"] {
+            Value::Null => None,
+            value => Some(
+                value
+                    .as_array()
+                    .and_then(|items| {
+                        items
+                            .iter()
+                            .map(|item| item.as_str().map(str::to_string))
+                            .collect()
+                    })
+                    .filter(|argv: &Vec<String>| !argv.is_empty())
+                    .ok_or_else(|| {
+                        Failure::new("EINVAL", "argv must be a non-empty array of strings")
+                    })?,
+            ),
+        };
+        Ok(PtyRequest {
+            argv,
+            cwd: json["cwd"]
+                .as_str()
+                .ok_or_else(|| Failure::new("EINVAL", "pty needs cwd"))?
+                .to_string(),
+            env: json["env"].as_object().cloned().unwrap_or_default(),
+            inherit_env: json["inheritEnv"].as_bool().unwrap_or(true),
+            cols: size(json, "cols")?,
+            rows: size(json, "rows")?,
+        })
+    }
+}
+
+fn find(terminals: &Terminals, json: &Value) -> Result<Arc<Terminal>, Failure> {
+    let target = json["target"]
+        .as_u64()
+        .ok_or_else(|| Failure::new("EINVAL", "missing field target"))?;
+    terminals
+        .lock()
+        .unwrap()
+        .get(&(target as u32))
+        .cloned()
+        .ok_or_else(|| Failure::new("EBADF", "no running terminal with this id"))
+}
+
+/// Queue input for the terminal; it is written in the order it arrives, without holding up other requests.
+pub fn write(terminals: &Terminals, json: &Value, payload: Vec<u8>) -> Result<Value, Failure> {
+    let terminal = find(terminals, json)?;
+    terminal
+        .input
+        .send(payload)
+        .map_err(|_| Failure::new("EBADF", "the terminal has ended"))?;
+    Ok(json!({}))
+}
+
+/// Change the terminal's size; the program receives `SIGWINCH`.
+pub fn resize(terminals: &Terminals, json: &Value) -> Result<Value, Failure> {
+    let terminal = find(terminals, json)?;
+    let (cols, rows) = (size(json, "cols")?, size(json, "rows")?);
+    #[cfg(unix)]
+    unix::set_size(&terminal.master, cols, rows)
+        .map_err(|error| Failure::io(&error, "ioctl", "pty"))?;
+    #[cfg(not(unix))]
+    let _ = (terminal, cols, rows);
+    Ok(json!({}))
+}
+
+#[cfg(not(unix))]
+pub fn run(
+    _id: u32,
+    _request: PtyRequest,
+    _output: &Arc<Output>,
+    _control: &Control,
+    _groups: &Groups,
+    _terminals: &Terminals,
+) -> Result<Value, Failure> {
+    Err(Failure::new(
+        "ENOTSUP",
+        "Terminals are not supported on this system yet",
+    ))
+}
+
+/// Run the program in a new terminal, sending a `started` event and then its output, until it exits.
+#[cfg(unix)]
+pub fn run(
+    id: u32,
+    request: PtyRequest,
+    output: &Arc<Output>,
+    control: &Control,
+    groups: &Groups,
+    terminals: &Terminals,
+) -> Result<Value, Failure> {
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::mpsc::{self, RecvTimeoutError};
+    use std::thread;
+    use std::time::{Duration, Instant};
+
+    let (sender, receiver) = mpsc::channel();
+    control.attach(sender.clone());
+    if control.aborted.load(Ordering::SeqCst) {
+        return Err(Failure::new("aborted", "aborted"));
+    }
+    if !std::path::Path::new(&request.cwd).exists() {
+        return Err(Failure::new(
+            "spawn_error",
+            format!("Working directory does not exist: {}", request.cwd),
+        ));
+    }
+    let (master, mut child) =
+        unix::spawn(&request).map_err(|error| Failure::new("spawn_error", error.to_string()))?;
+    let master = Arc::new(master);
+    let pid = child.id();
+    groups.lock().unwrap().insert(pid);
+    let (input, inputs) = mpsc::channel::<Vec<u8>>();
+    terminals.lock().unwrap().insert(
+        id,
+        Arc::new(Terminal {
+            input,
+            master: master.clone(),
+        }),
+    );
+    output.control(Frame::new(
+        EVENT,
+        id,
+        json!({ "kind": "started", "pid": pid }),
+    ));
+
+    let stop = Arc::new(AtomicBool::new(false));
+    let reader = {
+        let master = master.clone();
+        let sender = sender.clone();
+        let stop = stop.clone();
+        let output = output.clone();
+        thread::spawn(move || {
+            unix::read(&master, &stop, |bytes| {
+                let delivered = sender.send(Message::Data(0, bytes.to_vec())).is_ok();
+                // Like a full pipe: a program that writes faster than the link waits.
+                output.wait_for_room(&stop);
+                delivered
+            });
+            let _ = sender.send(Message::Eof(0));
+        })
+    };
+    {
+        let master = master.clone();
+        let stop = stop.clone();
+        thread::spawn(move || {
+            for bytes in inputs {
+                if !unix::write_all(&master, &bytes, &stop) {
+                    return;
+                }
+            }
+        });
+    }
+    let waiter = sender.clone();
+    thread::spawn(move || {
+        let _ = waiter.send(match child.wait() {
+            Ok(status) => Message::Exit(status),
+            Err(error) => Message::WaitFailed(error.to_string()),
+        });
+    });
+    drop(sender);
+
+    let mut status = None;
+    let mut ended = false;
+    let mut idle_until: Option<Instant> = None;
+    let mut aborted = false;
+    let mut killed = false;
+    let kill = |killed: &mut bool| {
+        if !*killed {
+            *killed = true;
+            kill_group(pid);
+        }
+    };
+    if control.killed.load(Ordering::SeqCst) {
+        kill(&mut killed);
+    }
+    loop {
+        if status.is_some() && ended {
+            break;
+        }
+        let wait = match idle_until {
+            Some(idle) => {
+                let now = Instant::now();
+                if now >= idle {
+                    // A background job may keep the terminal open after the program exited.
+                    break;
+                }
+                idle - now
+            }
+            None => Duration::from_secs(3600),
+        };
+        match receiver.recv_timeout(wait) {
+            Ok(Message::Data(_, bytes)) => {
+                output.bulk(
+                    Frame::with_payload(EVENT, id, json!({ "kind": "output" }), bytes),
+                    None,
+                );
+                if status.is_some() {
+                    idle_until = Some(Instant::now() + EXIT_GRACE);
+                }
+            }
+            Ok(Message::Eof(_)) => ended = true,
+            Ok(Message::Exit(exit)) => {
+                status = Some(Ok(exit));
+                idle_until = Some(Instant::now() + EXIT_GRACE);
+            }
+            Ok(Message::WaitFailed(error)) => {
+                status = Some(Err(error));
+                idle_until = Some(Instant::now());
+            }
+            Ok(Message::Cancel) => {
+                aborted = true;
+                kill(&mut killed);
+            }
+            Ok(Message::Kill) => kill(&mut killed),
+            Err(RecvTimeoutError::Timeout) => {}
+            Err(RecvTimeoutError::Disconnected) => break,
+        }
+    }
+    stop.store(true, Ordering::SeqCst);
+    // Ending the input channel ends the writer thread.
+    terminals.lock().unwrap().remove(&id);
+    groups.lock().unwrap().remove(&pid);
+    let _ = reader.join();
+    if aborted {
+        return Err(Failure::new("aborted", "aborted"));
+    }
+    match status {
+        Some(Ok(status)) => Ok(json!({ "exitCode": crate::sys::exit_code(status) })),
+        Some(Err(error)) => Err(Failure::new("spawn_error", error)),
+        None => Err(Failure::new(
+            "spawn_error",
+            "process did not report its exit",
+        )),
+    }
+}
+
+#[cfg(unix)]
+mod unix {
+    use super::PtyRequest;
+    use std::fs::File;
+    use std::io::{self, Read, Write};
+    use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
+    use std::os::unix::process::CommandExt;
+    use std::process::{Child, Command, Stdio};
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    use serde_json::Value;
+
+    fn check(result: libc::c_int) -> io::Result<libc::c_int> {
+        if result < 0 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(result)
+        }
+    }
+
+    fn winsize(cols: u16, rows: u16) -> libc::winsize {
+        libc::winsize {
+            ws_row: rows,
+            ws_col: cols,
+            ws_xpixel: 0,
+            ws_ypixel: 0,
+        }
+    }
+
+    pub fn set_size(master: &File, cols: u16, rows: u16) -> io::Result<()> {
+        let size = winsize(cols, rows);
+        // SAFETY: TIOCSWINSZ reads one winsize from a valid pointer.
+        check(unsafe { libc::ioctl(master.as_raw_fd(), libc::TIOCSWINSZ as _, &size) })?;
+        Ok(())
+    }
+
+    fn cloexec(fd: libc::c_int) -> io::Result<()> {
+        // SAFETY: fcntl on a descriptor this function's caller owns.
+        unsafe {
+            let flags = check(libc::fcntl(fd, libc::F_GETFD))?;
+            check(libc::fcntl(fd, libc::F_SETFD, flags | libc::FD_CLOEXEC))?;
+        }
+        Ok(())
+    }
+
+    /// The program to run: `argv`, or the user's shell as a login shell (`-` before its name, as `login` does).
+    fn command(request: &PtyRequest) -> Command {
+        match &request.argv {
+            Some(argv) => {
+                let mut command = Command::new(&argv[0]);
+                command.args(&argv[1..]);
+                command
+            }
+            None => {
+                let shell = request
+                    .env
+                    .get("SHELL")
+                    .and_then(Value::as_str)
+                    .map(str::to_string)
+                    .or_else(|| std::env::var("SHELL").ok())
+                    .filter(|shell| !shell.is_empty() && crate::sys::path_exists(shell))
+                    .unwrap_or_else(|| "/bin/sh".into());
+                let name = shell.rsplit('/').next().unwrap_or("sh").to_string();
+                let mut command = Command::new(&shell);
+                command.arg0(format!("-{name}"));
+                command
+            }
+        }
+    }
+
+    pub fn spawn(request: &PtyRequest) -> io::Result<(File, Child)> {
+        let mut master: libc::c_int = -1;
+        let mut slave: libc::c_int = -1;
+        let mut size = winsize(request.cols, request.rows);
+        // SAFETY: openpty writes two descriptors; the name and termios are not requested.
+        check(unsafe {
+            libc::openpty(
+                &mut master,
+                &mut slave,
+                std::ptr::null_mut(),
+                std::ptr::null_mut(),
+                &raw mut size,
+            )
+        })?;
+        // SAFETY: openpty returned two new descriptors this function now owns.
+        let (master, slave) =
+            unsafe { (OwnedFd::from_raw_fd(master), OwnedFd::from_raw_fd(slave)) };
+        cloexec(master.as_raw_fd())?;
+        cloexec(slave.as_raw_fd())?;
+        let slave = File::from(slave);
+        let mut command = command(request);
+        command
+            .current_dir(&request.cwd)
+            .stdin(Stdio::from(slave.try_clone()?))
+            .stdout(Stdio::from(slave.try_clone()?))
+            .stderr(Stdio::from(slave.try_clone()?));
+        if !request.inherit_env {
+            command.env_clear();
+        }
+        for (key, value) in &request.env {
+            if let Some(value) = value.as_str() {
+                command.env(key, value);
+            }
+        }
+        // SAFETY: only async-signal-safe calls between fork and exec.
+        unsafe {
+            command.pre_exec(|| {
+                // A new session whose controlling terminal is the pty (stdin), so job control and SIGHUP work; default
+                // signal handling and an empty mask, as for `exec`.
+                if libc::setsid() < 0 {
+                    return Err(io::Error::last_os_error());
+                }
+                if libc::ioctl(0, libc::TIOCSCTTY as _, 0) < 0 {
+                    return Err(io::Error::last_os_error());
+                }
+                for signal in 1..32 {
+                    if signal != libc::SIGKILL && signal != libc::SIGSTOP {
+                        libc::signal(signal, libc::SIG_DFL);
+                    }
+                }
+                let mut mask: libc::sigset_t = std::mem::zeroed();
+                libc::sigemptyset(&mut mask);
+                libc::sigprocmask(libc::SIG_SETMASK, &mask, std::ptr::null_mut());
+                Ok(())
+            });
+        }
+        let child = command.spawn()?;
+        // Only the program holds the terminal now, so reads end once it and its descendants close it.
+        drop(slave);
+        let master = File::from(master);
+        // SAFETY: fcntl on the descriptor owned above; reads and writes poll instead of blocking.
+        unsafe {
+            let flags = check(libc::fcntl(master.as_raw_fd(), libc::F_GETFL))?;
+            check(libc::fcntl(
+                master.as_raw_fd(),
+                libc::F_SETFL,
+                flags | libc::O_NONBLOCK,
+            ))?;
+        }
+        Ok((master, child))
+    }
+
+    fn poll(fd: libc::c_int, events: libc::c_short) -> io::Result<bool> {
+        let mut poll = libc::pollfd {
+            fd,
+            events,
+            revents: 0,
+        };
+        // SAFETY: one valid pollfd.
+        let ready = unsafe { libc::poll(&mut poll, 1, 100) };
+        if ready < 0 {
+            let error = io::Error::last_os_error();
+            return if error.kind() == io::ErrorKind::Interrupted {
+                Ok(false)
+            } else {
+                Err(error)
+            };
+        }
+        Ok(ready > 0)
+    }
+
+    /// Read until the terminal closes (Linux reports `EIO` once no program holds it), `on_data` returns false, or
+    /// `stop` is set.
+    pub fn read(master: &File, stop: &AtomicBool, mut on_data: impl FnMut(&[u8]) -> bool) {
+        let mut buffer = vec![0u8; 64 * 1024];
+        while !stop.load(Ordering::SeqCst) {
+            match poll(master.as_raw_fd(), libc::POLLIN) {
+                Ok(false) => continue,
+                Ok(true) => {}
+                Err(_) => return,
+            }
+            match (&*master).read(&mut buffer) {
+                Ok(0) => return,
+                Ok(read) => {
+                    if !on_data(&buffer[..read]) {
+                        return;
+                    }
+                }
+                Err(error)
+                    if matches!(
+                        error.kind(),
+                        io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                    ) => {}
+                Err(_) => return,
+            }
+        }
+    }
+
+    /// Write all of `bytes`, waiting while the program does not read its input; false once the terminal is gone.
+    pub fn write_all(master: &File, mut bytes: &[u8], stop: &AtomicBool) -> bool {
+        while !bytes.is_empty() {
+            if stop.load(Ordering::SeqCst) {
+                return false;
+            }
+            match (&*master).write(bytes) {
+                Ok(0) => return false,
+                Ok(written) => bytes = &bytes[written..],
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    if poll(master.as_raw_fd(), libc::POLLOUT).is_err() {
+                        return false;
+                    }
+                }
+                Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
+                Err(_) => return false,
+            }
+        }
+        true
+    }
+}

packages/env/docs/protocol.md

@@ -30,7 +30,7 @@ does not parse gets an `EINVAL` error; framing errors end the connection.
 | 1 | request | client → daemon | `json.op` names the operation |
 | 2 | result | daemon → client | success of request `id`; the request is finished |
 | 3 | error | daemon → client | failure of request `id`; the request is finished |
-| 4 | event | daemon → client | progress of a running request (`exec` output, `watch` changes), not finishing it |
+| 4 | event | daemon → client | progress of a running request (`exec` and `pty` output, `watch` changes), not finishing it |
 | 5 | cancel | client → daemon | cancel request `id`; see below |
 | 6 | ping | both | liveness; the receiver ignores it |
 
@@ -45,13 +45,14 @@ contents and output travel as raw payload bytes.
 ## Cancellation
 
 A cancel names one request. The daemon registers each request before running it, so a cancel sent right after its
-request is never lost. `exec` kills the command's process group and finishes with `aborted`; `{ mode: "kill" }` kills
+request is never lost. `exec` and `pty` kill the command's process group and finish with `aborted`; `{ mode: "kill" }` kills
 without aborting, so the command finishes with the killed process's status, as `cleanup()` does. `scanLines` stops
 between chunks with `aborted`. `watch` stops and finishes with `{}`. Other operations are short and finish normally.
 
 ## Scheduling
 
-File operations run on a fixed pool of worker threads; `exec` and `watch` run on their own threads. Chunk writes to one
+File operations run on a fixed pool of worker threads; `exec`, `pty` and `watch` run on their own threads. `ptyWrite`
+and `ptyResize` are answered in arrival order by the thread that reads frames, so keystrokes never swap. Chunk writes to one
 handle run in arrival order. Results, errors, pings and `watch` events are sent before queued command output, and
 results with payloads above 64 KiB queue behind output too, so a slow link delays bulk data, not replies. Output of a
 command without an output window waits while more than 4 MiB of output is unsent, which slows the command as a full
@@ -112,3 +113,17 @@ are `{ path, recursive, exclude: { hidden, names } }`. Once coverage is establis
 follows; then `{ kind: "change", paths }` or `{ kind: "change", overflow: true, mode: "polling" }` (native watching
 failed), or `{ kind: "error", code, message }`, after which nothing follows. The request lasts until cancelled.
 Failing to establish coverage is an error of the request.
+
+`pty { argv?, cwd, env, inheritEnv, cols, rows }` runs a program in a new pseudo-terminal of `cols` × `rows` cells
+(1 to 4096), as the leader of a new session with the terminal as its controlling terminal. Without `argv`, it runs
+`SHELL` (from `env`, else the daemon's environment, else `/bin/sh`) as a login shell (`-` before its name). The
+environment is built as for `exec`; set `TERM` in `env`. Once the program runs, an event `{ kind: "started", pid }`
+comes first; then `{ kind: "output" }` events carry the terminal's raw output bytes as payload (not decoded, unlike
+`exec`). Output waits while more than 4 MiB of output is unsent, like a full pipe. After the program exits, output is
+collected until the terminal closes or 100 ms pass without output. The result is `{ exitCode }`. Not supported on
+Windows yet (`ENOTSUP`).
+
+While a `pty` request runs, `ptyWrite { target }` writes its payload to the terminal's input, and
+`ptyResize { target, cols, rows }` changes its size (the program gets `SIGWINCH`); `target` is the id of the `pty`
+request. Both return `{}` at once; input is queued and written in order, waiting while the program does not read it.
+Once the terminal has ended, they fail with `EBADF`.

packages/env/src/index.ts

@@ -16,6 +16,7 @@ export {
 	forgetHostKey,
 	HostKeyChangedError,
 	HostKeyUnknownError,
+	localDaemon,
 	packagedDaemon,
 	type RemotePlatform,
 	type SshConnectOptions,
@@ -25,4 +26,5 @@ export {
 	sshArguments,
 	sshConnection,
 } from "./ssh.ts";
+export { openTerminal, RemoteTerminal, type TerminalOptions } from "./terminal.ts";
 export { RemoteWatcher, type RemoteWatchOptions } from "./watch.ts";

packages/env/src/ssh.ts

@@ -358,6 +358,15 @@ export function packagedDaemon(remote: Pick<RemotePlatform, "platform" | "arch">
 	return join(packageRoot, "bin", name, remote.platform === "windows" ? "pi-env.exe" : "pi-env");
 }
 
+/** The daemon this package ships for the machine it runs on, e.g. for terminals of local sessions. */
+export function localDaemon(): string {
+	const platform = process.platform === "win32" ? "windows" : process.platform;
+	if (platform !== "linux" && platform !== "android" && platform !== "darwin" && platform !== "windows") {
+		throw new Error(`No pi-env daemon for ${process.platform}`);
+	}
+	return packagedDaemon({ platform, arch: normalizeArch(process.arch) });
+}
+
 function quotePosix(text: string): string {
 	return `'${text.replaceAll("'", "'\\''")}'`;
 }

packages/env/src/terminal.ts

@@ -0,0 +1,110 @@
+import type { Connection, Json } from "./connection.ts";
+
+/** Input larger than this goes to the daemon in several writes; frames are at most 16 MiB. */
+const WRITE_CHUNK = 1024 * 1024;
+
+export interface TerminalOptions {
+	/** Working directory of the program; it must exist. */
+	cwd: string;
+	/** Program and arguments. Default: the user's shell (`SHELL` in `env` or the daemon's environment) as a login shell. */
+	argv?: readonly string[];
+	/** Added to the daemon's environment, or the whole environment with `inheritEnv: false`. Set `TERM` here. */
+	env?: Record<string, string>;
+	inheritEnv?: boolean;
+	cols: number;
+	rows: number;
+	/** The terminal's raw output, in order. */
+	onData: (bytes: Uint8Array) => void;
+}
+
+/**
+ * A program running in a pseudo-terminal of a pi-env daemon. It ends when the program exits, `close()` kills it, or
+ * the connection is lost (the daemon then kills it).
+ */
+export class RemoteTerminal {
+	readonly pid: number;
+	/** The exit code; rejects with the daemon's error, with code `aborted` after `close()`. */
+	readonly exited: Promise<number>;
+	readonly #connection: Connection;
+	readonly #id: number;
+	readonly #session: number;
+	readonly #abort: AbortController;
+
+	constructor(
+		connection: Connection,
+		id: number,
+		session: number,
+		pid: number,
+		exited: Promise<number>,
+		abort: AbortController,
+	) {
+		this.#connection = connection;
+		this.#id = id;
+		this.#session = session;
+		this.pid = pid;
+		this.exited = exited;
+		this.#abort = abort;
+	}
+
+	/** Send input; once the terminal has ended, input is dropped. */
+	write(data: Uint8Array | string): void {
+		const bytes = typeof data === "string" ? new TextEncoder().encode(data) : data;
+		for (let offset = 0; offset < bytes.length; offset += WRITE_CHUNK) {
+			this.#request("ptyWrite", {}, bytes.subarray(offset, offset + WRITE_CHUNK));
+		}
+	}
+
+	/** Change the size; the program receives `SIGWINCH`. */
+	resize(cols: number, rows: number): void {
+		this.#request("ptyResize", { cols, rows });
+	}
+
+	/** Kill the program and everything in its session. */
+	close(): void {
+		this.#abort.abort();
+	}
+
+	#request(op: string, json: Json, payload?: Uint8Array): void {
+		this.#connection
+			.request(op, { ...json, target: this.#id }, { session: this.#session, ...(payload ? { payload } : {}) })
+			// A terminal that ended or a lost connection: `exited` reports it.
+			.catch(() => {});
+	}
+}
+
+/** Start a program in a new pseudo-terminal of the daemon behind `connection`. */
+export function openTerminal(connection: Connection, options: TerminalOptions): Promise<RemoteTerminal> {
+	const abort = new AbortController();
+	const started = Promise.withResolvers<RemoteTerminal>();
+	let request: { id: number; session: number } | undefined;
+	const exited = connection
+		.request(
+			"pty",
+			{
+				cwd: options.cwd,
+				...(options.argv === undefined ? {} : { argv: options.argv }),
+				...(options.env === undefined ? {} : { env: options.env }),
+				...(options.inheritEnv === undefined ? {} : { inheritEnv: options.inheritEnv }),
+				cols: options.cols,
+				rows: options.rows,
+			},
+			{
+				signal: abort.signal,
+				onStart: (id, session) => {
+					request = { id, session };
+				},
+				onEvent: (json, payload) => {
+					if (json.kind === "output") options.onData(payload);
+					else if (json.kind === "started" && request) {
+						started.resolve(
+							new RemoteTerminal(connection, request.id, request.session, Number(json.pid), exited, abort),
+						);
+					}
+				},
+			},
+		)
+		.then((reply) => Number(reply.json.exitCode));
+	// Failing before it started rejects the open; afterwards, `exited` carries it.
+	exited.catch(started.reject);
+	return started.promise;
+}

packages/env/test/terminal.test.ts

@@ -0,0 +1,131 @@
+import { mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterAll, describe, expect, it } from "vitest";
+import { Connection, RemoteError } from "../src/connection.ts";
+import { openTerminal } from "../src/terminal.ts";
+import { daemon } from "./daemon.ts";
+
+const connections: Connection[] = [];
+const dirs: string[] = [];
+afterAll(() => {
+	for (const connection of connections) connection.close();
+	for (const dir of dirs) rmSync(dir, { recursive: true, force: true });
+});
+
+function connect(): { connection: Connection; cwd: string } {
+	const cwd = mkdtempSync(join(tmpdir(), "pi-env-terminal-"));
+	dirs.push(cwd);
+	const connection = new Connection({ command: [daemon] });
+	connections.push(connection);
+	return { connection, cwd };
+}
+
+/** Collects output and waits for text to appear in it. */
+function collector(): {
+	onData: (bytes: Uint8Array) => void;
+	bytes: () => Buffer;
+	waitFor: (text: string) => Promise<void>;
+} {
+	const chunks: Buffer[] = [];
+	const waiting: { text: string; resolve: () => void }[] = [];
+	const bytes = () => Buffer.concat(chunks);
+	return {
+		bytes,
+		onData: (data) => {
+			chunks.push(Buffer.from(data));
+			const all = bytes().toString("latin1");
+			for (const wait of waiting.splice(0)) {
+				if (all.includes(wait.text)) wait.resolve();
+				else waiting.push(wait);
+			}
+		},
+		waitFor: (text) =>
+			bytes().toString("latin1").includes(text)
+				? Promise.resolve()
+				: new Promise((resolve) => waiting.push({ text, resolve })),
+	};
+}
+
+describe.skipIf(process.platform === "win32")("openTerminal", () => {
+	it("runs a program in a terminal with input, resizing and its exit code", async () => {
+		const { connection, cwd } = connect();
+		const output = collector();
+		const terminal = await openTerminal(connection, {
+			cwd,
+			argv: ["/bin/sh"],
+			env: { PS1: "$ ", TERM: "xterm-256color" },
+			cols: 80,
+			rows: 24,
+			onData: output.onData,
+		});
+		expect(terminal.pid).toBeGreaterThan(0);
+		terminal.write("test -t 0 && echo IS-A-TTY; stty size; echo $TERM\n");
+		await output.waitFor("IS-A-TTY");
+		await output.waitFor("24 80");
+		await output.waitFor("xterm-256color");
+		terminal.resize(100, 30);
+		terminal.write("stty size\n");
+		await output.waitFor("30 100");
+		terminal.write("exit 3\n");
+		expect(await terminal.exited).toBe(3);
+	}, 20_000);
+
+	it("delivers raw bytes, not decoded text", async () => {
+		const { connection, cwd } = connect();
+		const output = collector();
+		const terminal = await openTerminal(connection, {
+			cwd,
+			argv: ["sh", "-c", "printf 'a\\377b'"],
+			cols: 80,
+			rows: 24,
+			onData: output.onData,
+		});
+		expect(await terminal.exited).toBe(0);
+		expect([...output.bytes()]).toEqual([0x61, 0xff, 0x62]);
+	});
+
+	it("keeps input in order", async () => {
+		const { connection, cwd } = connect();
+		const terminal = await openTerminal(connection, {
+			cwd,
+			argv: ["sh", "-c", "head -c 40000 > input.txt"],
+			cols: 80,
+			rows: 24,
+			onData: () => {},
+		});
+		let text = "";
+		for (let index = 0; index < 4000; index++) {
+			const piece = `${String(index).padStart(9, "0")}\n`;
+			text += piece;
+			terminal.write(piece);
+		}
+		expect(await terminal.exited).toBe(0);
+		expect(readFileSync(join(cwd, "input.txt"), "utf8")).toBe(text);
+	}, 20_000);
+
+	it("kills the program on close", async () => {
+		const { connection, cwd } = connect();
+		const terminal = await openTerminal(connection, {
+			cwd,
+			argv: ["sleep", "100"],
+			cols: 80,
+			rows: 24,
+			onData: () => {},
+		});
+		terminal.close();
+		await expect(terminal.exited).rejects.toMatchObject({ code: "aborted" });
+	});
+
+	it("fails to open in a missing directory", async () => {
+		const { connection, cwd } = connect();
+		await expect(
+			openTerminal(connection, { cwd: join(cwd, "missing"), cols: 80, rows: 24, onData: () => {} }),
+		).rejects.toBeInstanceOf(RemoteError);
+	});
+
+	it("rejects writes to a terminal that does not exist", async () => {
+		const { connection } = connect();
+		await expect(connection.request("ptyWrite", { target: 999 })).rejects.toMatchObject({ code: "EBADF" });
+	});
+});