diff --git a/packages/process-daemon/SPEC.md b/packages/process-daemon/SPEC.md index d4b88f08f1f9..aceaede1b2f7 100644 --- a/packages/process-daemon/SPEC.md +++ b/packages/process-daemon/SPEC.md @@ -107,6 +107,44 @@ when it changes; an OSC change hidden behind a user title publishes nothing. `list` returns terminals in creation order. +## Pipe Processes + +`Create` takes `mode: "pty" | "pipe"` (default `"pty"`). A pipe process runs +any command without a terminal; it is Unix-only for now, and only daemons that +list the `pipe` feature accept it. Older daemons ignore `mode` and would start +a PTY instead, so clients check `features` first. + +- `title`, `cols` and `rows` are optional for pipe processes; the title + defaults to the program's file name and the size is reported as `0x0`. +- The child gets its own process group, an empty signal mask with default + actions (the daemon blocks SIGTERM and SIGINT, and std keeps the mask across + exec), stdin on `/dev/null`, and one pipe for + both stdout and stderr, so their bytes stay in order. No `TERM` or + `COLORTERM` is added, and nothing parses the output: there is no emulator, + no terminal replies, no OSC title and no foreground process. +- Output goes through the same raw replay and subscriptions as a PTY, with the + same absolute offsets. A pipe has no controller or observer distinction for + writing, since it accepts no input. +- `write`, `resize`, `control`, `input`, `snapshot`, `read_rows` and `rename` + fail for pipe processes. `replay`, `subscribe` and `terminate` + work. +- Exit is published once both the exit status and the end of output are + known. A descendant that keeps the pipe open (`cmd &`) delays that by at + most one second after the child exits; output after that is not read. +- `Exited` (the lifecycle and the subscription response) carries `signal`, + the signal's name such as `"SIGTERM"`, when a pipe process was killed by + one; `exit_code` is then null. The field is left out when there is no + signal, so PTY exits serialize as before. +- An exited pipe process stays listed, as a zombie, until `terminate`. The + zombie keeps its PID and process group ID reserved, so `terminate` can kill + any descendants left in the group without reaching an unrelated process. + `terminate` sends SIGKILL to the group and to the child, reaps it, and never + waits on a descendant that left the group and still holds the pipe. + +`List` takes `mode` too. Without it, `list` returns PTYs only, so servers that +predate pipe processes never see them. Clients leave `mode` out of the request +when it is unset: older daemons reject a `list` with any extra field. + ## Failure Boundary Losing the owning server connection stops `opencode-pty` and its terminals. @@ -131,7 +169,7 @@ subscriptions, title updates, rename, controller takeover, terminate, and destru shutdown, ownership acquisition, and restart handoff preparation. `Pong` and the registration carry `features`, the daemon's additive -capabilities (currently `["rename"]`). Additive changes, such as a new request +capabilities (currently `["rename", "pipe"]` on Unix and `["rename"]` on Windows). Additive changes, such as a new request or field older clients can ignore, extend `features` and keep the protocol version; clients check `features` before sending a new request, since a daemon that cannot parse one drops the connection. Only breaking changes bump diff --git a/packages/process-daemon/src/client.rs b/packages/process-daemon/src/client.rs index ea7503d7c0e3..084363e1dbdb 100644 --- a/packages/process-daemon/src/client.rs +++ b/packages/process-daemon/src/client.rs @@ -13,7 +13,7 @@ use crate::protocol::{ AttachmentRole, Envelope, SubscriptionEvent, read_frame, read_subscription_event, write_frame, }; use crate::protocol::{PROTOCOL_VERSION, Request, Response}; -use crate::service::{CreateTerminal, TerminalId, TerminalInfo, TerminalRows}; +use crate::service::{CreateTerminal, ProcessMode, TerminalId, TerminalInfo, TerminalRows}; #[cfg(unix)] use crate::transport::Connection; @@ -165,6 +165,7 @@ impl TerminalClient { env: request.env, cols: request.cols, rows: request.rows, + mode: request.mode, })? { Response::Created { terminal } => Ok(terminal), response => unexpected(response), @@ -172,7 +173,7 @@ impl TerminalClient { } pub fn list(&self) -> Result> { - match self.request(Request::List)? { + match self.request(Request::List { mode: None })? { Response::Terminals { terminals } => Ok(terminals), response => unexpected(response), } @@ -524,8 +525,13 @@ fn watch(directory: &Path, id: TerminalId) -> Result<()> { io::stdout().flush()?; } SubscriptionEvent::Response(response) => match *response { - Response::Exited { exit_code, .. } => { - eprintln!("\nterminal exited: {exit_code:?}"); + Response::Exited { + exit_code, signal, .. + } => { + match signal { + Some(signal) => eprintln!("\nterminal killed by {signal}"), + None => eprintln!("\nterminal exited: {exit_code:?}"), + } return Ok(()); } Response::Resized { cols, rows, .. } => { @@ -588,6 +594,7 @@ fn play(directory: &Path) -> Result<()> { cwd: env::current_dir()?, cols: 100, rows: 30, + mode: ProcessMode::Pty, } }; let info = client.create(request)?; @@ -665,6 +672,7 @@ fn play(directory: &Path) -> Result<()> { cwd: env::current_dir()?, cols: 100, rows: 30, + mode: ProcessMode::Pty, })?; active = Some(info.id); thread::sleep(Duration::from_millis(200)); diff --git a/packages/process-daemon/src/daemon/server.rs b/packages/process-daemon/src/daemon/server.rs index d54b4d1b3c60..f1686ab04b48 100644 --- a/packages/process-daemon/src/daemon/server.rs +++ b/packages/process-daemon/src/daemon/server.rs @@ -503,9 +503,11 @@ fn stream_subscription( }, StreamEvent::Exited { exit_code, + signal, final_offset, } => Response::Exited { exit_code, + signal, final_offset, }, StreamEvent::ControllerChanged { @@ -556,6 +558,7 @@ fn dispatch( env, cols, rows, + mode, } => Response::Created { terminal: service.create(CreateTerminal { program, @@ -566,10 +569,11 @@ fn dispatch( env, cols, rows, + mode, })?, }, - Request::List => Response::Terminals { - terminals: service.list()?, + Request::List { mode } => Response::Terminals { + terminals: service.list_mode(mode.unwrap_or_default())?, }, Request::Write { id, diff --git a/packages/process-daemon/src/protocol.rs b/packages/process-daemon/src/protocol.rs index 80e5c0be537a..64e16990a8ca 100644 --- a/packages/process-daemon/src/protocol.rs +++ b/packages/process-daemon/src/protocol.rs @@ -4,10 +4,14 @@ use std::path::PathBuf; use anyhow::{Context, Result, bail}; use serde::{Deserialize, Serialize}; -use crate::service::{TerminalId, TerminalInfo}; +use crate::service::{ProcessMode, TerminalId, TerminalInfo}; pub const PROTOCOL_VERSION: u32 = 7; /// Additive capabilities of this daemon. Breaking changes bump `PROTOCOL_VERSION`. +/// `pipe` (non-PTY processes) is Unix-only for now. +#[cfg(unix)] +pub const FEATURES: &[&str] = &["rename", "pipe"]; +#[cfg(not(unix))] pub const FEATURES: &[&str] = &["rename"]; pub fn features() -> Vec { @@ -38,17 +42,31 @@ pub enum Request { ticket: Option, }, PrepareHandoff, + /// Old daemons ignore `mode` and would start a PTY, so clients send + /// `"pipe"` only to a daemon that lists the `pipe` feature. Create { program: String, args: Vec, cwd: PathBuf, + #[serde(default)] title: String, group_id: String, env: std::collections::HashMap, + /// Terminal size; required for `pty`, ignored for `pipe`. + #[serde(default)] cols: u16, + #[serde(default)] rows: u16, + #[serde(default, skip_serializing_if = "ProcessMode::is_pty")] + mode: ProcessMode, + }, + /// Without `mode`, lists PTYs only, so older servers never see pipe processes. + /// `mode` is left out of the JSON when unset: older daemons reject a `list` + /// with any extra key. + List { + #[serde(default, skip_serializing_if = "Option::is_none")] + mode: Option, }, - List, Write { id: TerminalId, attachment_id: Option, @@ -161,6 +179,9 @@ pub enum Response { }, Exited { exit_code: Option, + /// Signal name such as `SIGTERM` when a pipe process was killed by one. + #[serde(default, skip_serializing_if = "Option::is_none")] + signal: Option, final_offset: u64, }, ControllerChanged { @@ -335,4 +356,82 @@ mod tests { serde_json::Value::Null ); } + + #[test] + fn pipe_fields_keep_the_wire_compatible_with_older_peers() { + use crate::service::{ProcessMode, TerminalLifecycle}; + + // An older daemon rejects `list` with any extra key, and an older + // client sends a bare `list` and a `create` with no mode. + assert_eq!( + serde_json::to_value(Request::List { mode: None }).unwrap(), + serde_json::json!({"op": "list"}) + ); + assert!(matches!( + serde_json::from_str::(r#"{"op":"list"}"#).unwrap(), + Request::List { mode: None } + )); + assert!(matches!( + serde_json::from_str::(r#"{"op":"list","mode":"pipe"}"#).unwrap(), + Request::List { + mode: Some(ProcessMode::Pipe) + } + )); + let create = serde_json::json!({ + "op": "create", "program": "sh", "args": [], "cwd": "/", + "title": "t", "group_id": "g", "env": {}, "cols": 80, "rows": 24, + }); + assert!(matches!( + serde_json::from_value::(create.clone()).unwrap(), + Request::Create { + mode: ProcessMode::Pty, + cols: 80, + .. + } + )); + // A PTY create serializes exactly as before. + assert_eq!( + serde_json::to_value(serde_json::from_value::(create.clone()).unwrap()) + .unwrap(), + create + ); + assert!(matches!( + serde_json::from_str::( + r#"{"op":"create","program":"sh","args":[],"cwd":"/","group_id":"g","env":{},"mode":"pipe"}"# + ) + .unwrap(), + Request::Create { + mode: ProcessMode::Pipe, + cols: 0, + rows: 0, + .. + } + )); + // Exits without a signal look exactly as before. + assert_eq!( + serde_json::to_value(Response::Exited { + exit_code: Some(1), + signal: None, + final_offset: 2, + }) + .unwrap(), + serde_json::json!({"type": "exited", "exit_code": 1, "final_offset": 2}) + ); + assert_eq!( + serde_json::to_value(TerminalLifecycle::Exited { + exit_code: Some(1), + signal: None, + }) + .unwrap(), + serde_json::json!({"status": "exited", "exit_code": 1}) + ); + assert_eq!( + serde_json::from_str::(r#"{"status":"exited","exit_code":null}"#) + .unwrap(), + TerminalLifecycle::Exited { + exit_code: None, + signal: None, + } + ); + } } diff --git a/packages/process-daemon/src/service.rs b/packages/process-daemon/src/service.rs index bea34d7fd8f1..e3873779d1ff 100644 --- a/packages/process-daemon/src/service.rs +++ b/packages/process-daemon/src/service.rs @@ -17,6 +17,8 @@ use serde::{Deserialize, Serialize}; use crate::ghostty::{Format, Terminal, TerminalOptions}; use crate::protocol::AttachmentRole; +#[cfg(unix)] +mod pipe; #[cfg(windows)] mod windows; @@ -33,17 +35,43 @@ const MAX_ROWS_BYTES: usize = 1024 * 1024; pub type TerminalId = u64; +/// How a process is attached. `Pty` runs under a pseudo-terminal parsed by +/// libghostty; `Pipe` runs with stdin closed and stdout and stderr merged into +/// one pipe, and supports only output, replay, subscription and termination. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ProcessMode { + #[default] + Pty, + Pipe, +} + +impl ProcessMode { + pub fn is_pty(&self) -> bool { + *self == Self::Pty + } +} + #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)] #[serde(rename_all = "snake_case", tag = "status")] pub enum TerminalLifecycle { Running, - Exited { exit_code: Option }, - Failed { message: String }, + Exited { + exit_code: Option, + /// Signal name such as `SIGTERM` when a pipe process was killed by one. + #[serde(default, skip_serializing_if = "Option::is_none")] + signal: Option, + }, + Failed { + message: String, + }, } #[derive(Clone, Debug, Serialize, Deserialize)] pub struct TerminalInfo { pub id: TerminalId, + #[serde(default)] + pub mode: ProcessMode, pub pid: Option, /// Effective title: the user title, else the latest non-empty OSC title, /// else the title given at creation. @@ -102,6 +130,7 @@ pub enum StreamEvent { }, Exited { exit_code: Option, + signal: Option, final_offset: u64, }, ControllerChanged { @@ -149,6 +178,7 @@ pub struct CreateTerminal { pub env: HashMap, pub cols: u16, pub rows: u16, + pub mode: ProcessMode, } impl CreateTerminal { @@ -168,6 +198,7 @@ impl CreateTerminal { env: HashMap::new(), cols: DEFAULT_COLS, rows: DEFAULT_ROWS, + mode: ProcessMode::Pty, }) } } @@ -194,7 +225,13 @@ impl TerminalService { } pub fn create(&self, request: CreateTerminal) -> Result { - validate_size(request.cols, request.rows)?; + match request.mode { + ProcessMode::Pty => validate_size(request.cols, request.rows)?, + #[cfg(not(unix))] + ProcessMode::Pipe => bail!("pipe processes are not supported on this platform"), + #[cfg(unix)] + ProcessMode::Pipe => {} + } if request.program.trim().is_empty() { bail!("terminal program cannot be empty"); } @@ -222,12 +259,20 @@ impl TerminalService { Ok(info) } + /// Lists PTYs; see [`Self::list_mode`] for pipe processes. pub fn list(&self) -> Result> { + self.list_mode(ProcessMode::Pty) + } + + pub fn list_mode(&self, mode: ProcessMode) -> Result> { + // Filter before refreshing, so a busy actor of the other mode never + // delays this listing. let mut handles = self .terminals .lock() .map_err(|_| anyhow!("terminal registry lock poisoned"))? .values() + .filter(|terminal| terminal.mode == mode) .cloned() .collect::>(); handles.sort_by_key(|terminal| terminal.order); @@ -236,7 +281,9 @@ impl TerminalService { // Windows has no foreground query; do not wait behind actor // backpressure just to read its already-cached metadata. #[cfg(not(windows))] - terminal.request(|reply| ActorMessage::RefreshForegroundProcess { reply })?; + if terminal.mode.is_pty() { + terminal.request(|reply| ActorMessage::RefreshForegroundProcess { reply })?; + } terminals.push(terminal.info()); } Ok(terminals) @@ -255,7 +302,7 @@ impl TerminalService { if bytes.len() > MAX_INPUT_BYTES { bail!("input exceeds {MAX_INPUT_BYTES} byte limit"); } - self.get(id)?.request(|reply| ActorMessage::Write { + self.pty(id)?.request(|reply| ActorMessage::Write { attachment_id, bytes, reply, @@ -273,8 +320,9 @@ impl TerminalService { cols: u16, rows: u16, ) -> Result<()> { + let terminal = self.pty(id)?; validate_size(cols, rows)?; - self.get(id)?.request(|reply| ActorMessage::Resize { + terminal.request(|reply| ActorMessage::Resize { attachment_id, cols, rows, @@ -289,8 +337,9 @@ impl TerminalService { cols: u16, rows: u16, ) -> Result<()> { + let terminal = self.pty(id)?; validate_size(cols, rows)?; - self.get(id)?.request(|reply| ActorMessage::Control { + terminal.request(|reply| ActorMessage::Control { attachment_id, cols, rows, @@ -307,11 +356,12 @@ impl TerminalService { rows: u16, bytes: Vec, ) -> Result<()> { + let terminal = self.pty(id)?; validate_size(cols, rows)?; if bytes.len() > MAX_INPUT_BYTES { bail!("input exceeds {MAX_INPUT_BYTES} byte limit"); } - self.get(id)?.request(|reply| ActorMessage::Control { + terminal.request(|reply| ActorMessage::Control { attachment_id, cols, rows, @@ -321,19 +371,19 @@ impl TerminalService { } pub fn snapshot(&self, id: TerminalId) -> Result { - self.get(id)? + self.pty(id)? .request(|reply| ActorMessage::Snapshot { reply }) } pub fn read_rows(&self, id: TerminalId, rows: Option) -> Result { - self.get(id)? + self.pty(id)? .request(|reply| ActorMessage::ReadRows { rows, reply }) } /// Sets or clears (`None` or empty) the user title, which takes precedence /// over titles set by the terminal's programs. pub fn rename(&self, id: TerminalId, title: Option) -> Result<()> { - self.get(id)? + self.pty(id)? .request(|reply| ActorMessage::Rename { title, reply }) } @@ -385,6 +435,15 @@ impl TerminalService { terminal.shutdown() } + /// Like `get`, but for requests that need a terminal, which pipe processes lack. + fn pty(&self, id: TerminalId) -> Result> { + let terminal = self.get(id)?; + if !terminal.mode.is_pty() { + bail!("terminal {id} is a pipe process, which has no terminal"); + } + Ok(terminal) + } + fn get(&self, id: TerminalId) -> Result> { self.terminals .lock() @@ -426,21 +485,111 @@ impl Drop for TerminalService { struct TerminalHandle { /// Creation sequence number; orders `list` results. order: u64, + mode: ProcessMode, actor_tx: Sender, /// Disconnects when the actor thread exits; never carries a message. actor_done: Receiver<()>, shutdown_tx: Mutex>>, killer: Mutex>, + /// Dropped at shutdown, before the joins, to cancel a pipe reader that a + /// surviving descendant would otherwise keep blocked. + cancel: Mutex>>, + /// Runs after the joins; reaps a pipe process kept as a zombie until now. + reap: Mutex>>, info: Arc>, joins: Mutex>>>, } +/// The actor thread and the channels other workers share with it. +struct ActorStart { + actor_tx: Sender, + actor_done: Receiver<()>, + shutdown_tx: Sender<()>, + shutdown: Receiver<()>, + thread: JoinHandle<()>, +} + +/// Starts the actor and waits for it to initialize. On error the actor has +/// stopped; the caller still owns, and must stop, the child. +fn start_actor( + id: TerminalId, + mode: ProcessMode, + master: ActorMaster, + writes: Sender, + info: &Arc>, + replay_capacity: usize, + (cols, rows): (u16, u16), +) -> Result { + let (actor_tx, actor_rx) = bounded(ACTOR_QUEUE_CAPACITY); + // Disconnecting this channel wakes every cancellable queue operation, + // even when the actor cannot consume an ordinary shutdown message. + let (shutdown_tx, shutdown) = bounded(0); + let (init_tx, init_rx) = std_mpsc::sync_channel(1); + let actor_info = Arc::clone(info); + let actor_shutdown = shutdown.clone(); + let (actor_done_tx, actor_done) = bounded::<()>(0); + let thread = thread::Builder::new() + .name(format!("opencode-pty-actor-{id}")) + .spawn(move || { + let _actor_done = actor_done_tx; + let result = run_actor(ActorConfig { + mode, + master, + messages: actor_rx, + writes, + shutdown: actor_shutdown, + info: actor_info, + replay_capacity, + cols, + rows, + init_tx, + }); + if let Err(error) = result { + eprintln!("terminal {id} actor failed: {error:#}"); + } + }) + .context("failed to spawn terminal actor")?; + let initialized = init_rx.recv(); + match initialized + .map_err(|_| anyhow!("terminal actor stopped during initialization")) + .and_then(|result| result.map_err(|error| anyhow!(error))) + { + Ok(()) => Ok(ActorStart { + actor_tx, + actor_done, + shutdown_tx, + shutdown, + thread, + }), + Err(error) => { + drop(shutdown_tx); + let _ = thread.join(); + Err(error) + } + } +} + impl TerminalHandle { fn spawn( id: TerminalId, order: u64, request: CreateTerminal, replay_capacity: usize, + ) -> Result { + match request.mode { + ProcessMode::Pty => Self::spawn_pty(id, order, request, replay_capacity), + #[cfg(unix)] + ProcessMode::Pipe => Self::spawn_pipe(id, order, request, replay_capacity), + #[cfg(not(unix))] + ProcessMode::Pipe => bail!("pipe processes are not supported on this platform"), + } + } + + fn spawn_pty( + id: TerminalId, + order: u64, + request: CreateTerminal, + replay_capacity: usize, ) -> Result { let pty_system = native_pty_system(); let pair = pty_system @@ -477,82 +626,41 @@ impl TerminalHandle { .context("failed to take PTY writer")?; drop(pair.slave); - let info = Arc::new(RwLock::new(TerminalInfo { - id, - pid, - title: request.title, - user_title: None, - foreground_process: None, - group_id: request.group_id, - command: std::iter::once(request.program) - .chain(request.args) - .collect(), - cwd: request.cwd, - cols: request.cols, - rows: request.rows, - lifecycle: TerminalLifecycle::Running, - output_head: 0, - output_tail: 0, - })); + let info = new_info(id, ProcessMode::Pty, pid, &request); - let (actor_tx, actor_rx) = bounded(ACTOR_QUEUE_CAPACITY); let (writer_tx, writer_rx) = bounded(WRITER_QUEUE_CAPACITY); - // Disconnecting this channel wakes every cancellable queue operation, - // even when the actor cannot consume an ordinary shutdown message. - let (shutdown_tx, shutdown) = bounded(0); #[cfg(windows)] let (close_tx, close_rx) = bounded(1); - let (init_tx, init_rx) = std_mpsc::sync_channel(1); - - let actor_info = Arc::clone(&info); - let actor_writer_tx = writer_tx.clone(); - let actor_shutdown = shutdown.clone(); - let (actor_done_tx, actor_done) = bounded::<()>(0); - let actor_thread = thread::Builder::new() - .name(format!("opencode-pty-actor-{id}")) - .spawn(move || { - let _actor_done = actor_done_tx; - let result = run_actor(ActorConfig { - master: ActorMaster { - pty: Some(pair.master), - #[cfg(windows)] - close_tx, - }, - messages: actor_rx, - writes: actor_writer_tx, - shutdown: actor_shutdown, - info: actor_info, - replay_capacity, - cols: request.cols, - rows: request.rows, - init_tx, - }); - if let Err(error) = result { - eprintln!("terminal {id} actor failed: {error:#}"); - } - }) - .context("failed to spawn terminal actor")?; - - match init_rx - .recv() - .context("terminal actor stopped during initialization")? - { - Ok(()) => {} + let actor = start_actor( + id, + ProcessMode::Pty, + ActorMaster { + pty: Some(pair.master), + #[cfg(windows)] + close_tx, + }, + writer_tx, + &info, + replay_capacity, + (request.cols, request.rows), + ); + let actor = match actor { + Ok(actor) => actor, Err(error) => { let mut killer = killer; let _ = killer.kill(); - let _ = actor_thread.join(); - bail!(error); + return Err(error); } - } + }; - let writer_actor_tx = actor_tx.clone(); + let writer_actor_tx = actor.actor_tx.clone(); + let writer_shutdown = actor.shutdown.clone(); let writer_thread = thread::Builder::new() .name(format!("opencode-pty-writer-{id}")) - .spawn(move || run_writer(writer, writer_rx, writer_actor_tx, shutdown)) + .spawn(move || run_writer(writer, writer_rx, writer_actor_tx, writer_shutdown)) .context("failed to spawn terminal writer")?; - let reader_actor_tx = actor_tx.clone(); + let reader_actor_tx = actor.actor_tx.clone(); let reader_thread = thread::Builder::new() .name(format!("opencode-pty-reader-{id}")) .spawn(move || { @@ -594,7 +702,7 @@ impl TerminalHandle { }) .context("failed to spawn terminal reader")?; - let wait_actor_tx = actor_tx.clone(); + let wait_actor_tx = actor.actor_tx.clone(); let wait_thread = thread::Builder::new() .name(format!("opencode-pty-wait-{id}")) .spawn(move || { @@ -613,13 +721,16 @@ impl TerminalHandle { Ok(Self { order, - actor_tx, - actor_done, - shutdown_tx: Mutex::new(Some(shutdown_tx)), + mode: ProcessMode::Pty, + actor_tx: actor.actor_tx, + actor_done: actor.actor_done, + shutdown_tx: Mutex::new(Some(actor.shutdown_tx)), killer: Mutex::new(killer), + cancel: Mutex::new(None), + reap: Mutex::new(None), info, joins: Mutex::new(Some(vec![ - actor_thread, + actor.thread, writer_thread, reader_thread, wait_thread, @@ -659,13 +770,60 @@ impl TerminalHandle { if let Ok(mut killer) = self.killer.lock() { let _ = killer.kill(); } + self.cancel + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take(); for join in handles { let _ = join.join(); } + if let Some(reap) = self + .reap + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .take() + { + reap(); + } Ok(()) } } +fn new_info( + id: TerminalId, + mode: ProcessMode, + pid: Option, + request: &CreateTerminal, +) -> Arc> { + let title = match mode { + ProcessMode::Pipe if request.title.is_empty() => PathBuf::from(&request.program) + .file_name() + .and_then(|value| value.to_str()) + .unwrap_or(&request.program) + .to_string(), + _ => request.title.clone(), + }; + Arc::new(RwLock::new(TerminalInfo { + id, + mode, + pid, + title, + user_title: None, + foreground_process: None, + group_id: request.group_id.clone(), + command: std::iter::once(request.program.clone()) + .chain(request.args.iter().cloned()) + .collect(), + cwd: request.cwd.clone(), + // Pipe processes have no terminal size. + cols: if mode.is_pty() { request.cols } else { 0 }, + rows: if mode.is_pty() { request.rows } else { 0 }, + lifecycle: TerminalLifecycle::Running, + output_head: 0, + output_tail: 0, + })) +} + type Reply = Sender; /// A request accepted into the actor queue may never be handled: shutdown @@ -687,6 +845,9 @@ enum ActorMessage { ReaderFailed(String), WriterFailed(String), ChildExited(Result, String>), + /// A pipe process ended by a signal; the value is the signal's name. + #[cfg(unix)] + ChildKilled(String), #[cfg(not(windows))] RefreshForegroundProcess { reply: Reply>, @@ -800,6 +961,7 @@ impl Drop for ActorMaster { } struct ActorConfig { + mode: ProcessMode, master: ActorMaster, messages: Receiver, writes: Sender, @@ -813,6 +975,7 @@ struct ActorConfig { fn run_actor(config: ActorConfig) -> Result<()> { let ActorConfig { + mode, master, messages, writes, @@ -825,11 +988,15 @@ fn run_actor(config: ActorConfig) -> Result<()> { } = config; #[cfg(windows)] let mut master = master; - let mut terminal = Terminal::new(TerminalOptions { - cols, - rows, - max_scrollback: replay_capacity, - })?; + // Pipe processes have no terminal: no emulator, replies, title or size. + let mut terminal = match mode { + ProcessMode::Pty => Some(Terminal::new(TerminalOptions { + cols, + rows, + max_scrollback: replay_capacity, + })?), + ProcessMode::Pipe => None, + }; let create_title = info .read() @@ -841,7 +1008,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { let mut subscribers = HashMap::::new(); let mut controller = None::; let mut controller_generation = 0_u64; - let mut child_exit = None::, String>>; + let mut child_exit = None::, Option), String>>; let mut reader_eof = false; let mut final_exit = None::; let mut pending_message = None; @@ -865,18 +1032,20 @@ fn run_actor(config: ActorConfig) -> Result<()> { } Ok(ActorMessage::Output(bytes)) => { let (start, end) = replay.append(&bytes); - terminal.vt_write(&bytes); - // Once the Windows master is handed off for closing, the writer - // may already have stopped; replies to the closing console are - // discarded so final output still reaches the real EOF. - #[cfg(windows)] - let closing = master.get().is_err(); - #[cfg(not(windows))] - let closing = false; - if closing { - terminal.take_writes(); - } else if let Err(error) = forward_replies(&mut terminal, &writes, &shutdown) { - break Err(error); + if let Some(terminal) = terminal.as_mut() { + terminal.vt_write(&bytes); + // Once the Windows master is handed off for closing, the writer + // may already have stopped; replies to the closing console are + // discarded so final output still reaches the real EOF. + #[cfg(windows)] + let closing = master.get().is_err(); + #[cfg(not(windows))] + let closing = false; + if closing { + terminal.take_writes(); + } else if let Err(error) = forward_replies(terminal, &writes, &shutdown) { + break Err(error); + } } update_offsets(&info, &replay); broadcast( @@ -886,7 +1055,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { &mut controller_generation, StreamEvent::Output { start, end, bytes }, ); - if let Some(title) = terminal.take_title() { + if let Some(title) = terminal.as_mut().and_then(Terminal::take_title) { osc_title = Some(title).filter(|title| !title.is_empty()); if let Ok(Some(title)) = refresh_title(&info, None, osc_title.as_deref(), &create_title) @@ -922,6 +1091,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { }) => { let result = (|| { authorize_controller(&controller, attachment_id.as_deref())?; + let terminal = emulator(&mut terminal)?; master.get()?.resize(PtySize { rows, cols, @@ -929,14 +1099,14 @@ fn run_actor(config: ActorConfig) -> Result<()> { pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes, &shutdown)?; + forward_replies(terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; value.rows = rows; } let generation = controller_generation; - let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); + let checkpoint = format_terminal(terminal, Format::Vt)?.into_bytes(); broadcast( &shutdown, &mut subscribers, @@ -993,6 +1163,7 @@ fn run_actor(config: ActorConfig) -> Result<()> { .map(|value| value.cols != cols || value.rows != rows) .unwrap_or(true); if changed { + let terminal = emulator(&mut terminal)?; master.get()?.resize(PtySize { rows, cols, @@ -1000,14 +1171,14 @@ fn run_actor(config: ActorConfig) -> Result<()> { pixel_height: 0, })?; let resized = terminal.resize(cols, rows); - forward_replies(&mut terminal, &writes, &shutdown)?; + forward_replies(terminal, &writes, &shutdown)?; resized?; if let Ok(mut value) = info.write() { value.cols = cols; value.rows = rows; } let generation = controller_generation; - let checkpoint = format_terminal(&terminal, Format::Vt)?.into_bytes(); + let checkpoint = format_terminal(terminal, Format::Vt)?.into_bytes(); broadcast( &shutdown, &mut subscribers, @@ -1031,12 +1202,15 @@ fn run_actor(config: ActorConfig) -> Result<()> { let _ = reply.send(result); } Ok(ActorMessage::Snapshot { reply }) => { - let _ = reply.send(make_snapshot(&terminal, &info)); + let _ = reply.send( + emulator(&mut terminal).and_then(|terminal| make_snapshot(terminal, &info)), + ); } Ok(ActorMessage::ReadRows { rows, reply }) => { let result = (|| { + let terminal = emulator(&mut terminal)?; Ok(TerminalRows { - lines: format_rows(&terminal, rows)?, + lines: format_rows(terminal, rows)?, terminal: info .read() .map_err(|_| anyhow!("terminal info lock poisoned"))? @@ -1166,8 +1340,12 @@ fn run_actor(config: ActorConfig) -> Result<()> { ); } } + #[cfg(unix)] + Ok(ActorMessage::ChildKilled(signal)) => { + child_exit = Some(Ok((None, Some(signal)))); + } Ok(ActorMessage::ChildExited(result)) => { - child_exit = Some(result); + child_exit = Some(result.map(|exit_code| (exit_code, None))); // ConPTY keeps its output pipe open until HPCON is closed. // The wait worker closes it while this actor and the reader // keep processing output, including the real final EOF. @@ -1195,15 +1373,16 @@ fn run_actor(config: ActorConfig) -> Result<()> { && final_exit.is_none() && let Some(result) = child_exit.take() { - let exit_code = result.as_ref().ok().copied().flatten(); + let (exit_code, signal) = result.clone().unwrap_or_default(); if let Ok(mut value) = info.write() { value.lifecycle = match result { - Ok(exit_code) => TerminalLifecycle::Exited { exit_code }, + Ok((exit_code, signal)) => TerminalLifecycle::Exited { exit_code, signal }, Err(message) => TerminalLifecycle::Failed { message }, }; } let event = StreamEvent::Exited { exit_code, + signal, final_offset: replay.tail, }; broadcast( @@ -1233,6 +1412,12 @@ fn run_actor(config: ActorConfig) -> Result<()> { } } +fn emulator(terminal: &mut Option) -> Result<&mut Terminal> { + terminal + .as_mut() + .ok_or_else(|| anyhow!("pipe processes have no terminal")) +} + fn receive_actor_message( messages: &Receiver, pending: &mut Option, @@ -1736,6 +1921,7 @@ mod tests { fn test_info() -> TerminalInfo { TerminalInfo { id: 1, + mode: ProcessMode::Pty, pid: None, title: "test".into(), user_title: None, @@ -1772,6 +1958,7 @@ mod tests { let actor = thread::spawn(move || { let _actor_done = actor_done_tx; run_actor(ActorConfig { + mode: ProcessMode::Pty, master: ActorMaster { pty: Some(Box::new(TestMaster)), #[cfg(windows)] @@ -1792,10 +1979,13 @@ mod tests { ( TerminalHandle { order: 0, + mode: ProcessMode::Pty, actor_tx, actor_done, shutdown_tx: Mutex::new(Some(shutdown_tx)), killer: Mutex::new(Box::new(TestKiller)), + cancel: Mutex::new(None), + reap: Mutex::new(None), info, joins: Mutex::new(Some(vec![ actor, @@ -1859,7 +2049,8 @@ mod tests { assert_eq!( snapshot.info.lifecycle, TerminalLifecycle::Exited { - exit_code: Some(23) + exit_code: Some(23), + signal: None, } ); if !eof_first { @@ -1926,6 +2117,7 @@ mod tests { events.as_slice(), [StreamEvent::Exited { exit_code: Some(23), + signal: None, final_offset: 5 }] )); diff --git a/packages/process-daemon/src/service/pipe.rs b/packages/process-daemon/src/service/pipe.rs new file mode 100644 index 000000000000..4af60ccaf3f4 --- /dev/null +++ b/packages/process-daemon/src/service/pipe.rs @@ -0,0 +1,444 @@ +//! Non-PTY ("pipe") processes on Unix. +//! +//! The child runs in its own process group with stdin on `/dev/null` and +//! stdout and stderr sharing one pipe, so their bytes keep their order. Three +//! choices keep the daemon safe from what the child leaves behind: +//! +//! - The waiter observes the exit with `WNOWAIT` and leaves the child as a +//! zombie until shutdown. While the zombie exists, its PID and process group +//! ID cannot be reused, so killing the group can never hit an unrelated +//! process, even long after the exit. +//! - A descendant can keep the pipe open after the child exits (`cmd &`, build +//! servers). The reader stops [`DRAIN_AFTER_EXIT`] after the exit instead of +//! waiting for an EOF that may never come. +//! - The reader polls a cancel pipe too, so shutdown never waits on a pipe that +//! a descendant which left the process group still holds open. + +use std::io::{self, PipeReader, PipeWriter, Read}; +use std::os::fd::AsFd; +use std::os::unix::process::CommandExt; +use std::process::{Command, Stdio}; +use std::sync::{Arc, Mutex, PoisonError, RwLock}; +use std::thread; +use std::time::{Duration, Instant}; + +use anyhow::{Context, Result}; +use crossbeam_channel::{Sender, bounded}; +use nix::errno::Errno; +use nix::poll::{PollFd, PollFlags, PollTimeout, poll}; +use portable_pty::ChildKiller; + +use super::{ + ActorMaster, ActorMessage, CreateTerminal, ProcessMode, TerminalHandle, TerminalId, new_info, + start_actor, +}; + +/// How long output is still read after the child exits. +const DRAIN_AFTER_EXIT: Duration = Duration::from_secs(1); + +/// Serializes pipe creation against `spawn`. macOS has no `pipe2`, so std +/// creates a pipe and then marks it close-on-exec; a fork in between would +/// leak the pipe into another child, where it could hold output or cancel +/// pipes open. Pipes are created under the read lock, children spawned under +/// the write lock. +static SPAWN_LOCK: RwLock<()> = RwLock::new(()); + +struct Pipes { + output: PipeReader, + output_writer: PipeWriter, + stderr: PipeWriter, + cancel: PipeReader, + cancel_writer: PipeWriter, + exited: PipeReader, + exited_writer: PipeWriter, +} + +impl Pipes { + fn new() -> Result { + let _guard = SPAWN_LOCK.read().unwrap_or_else(PoisonError::into_inner); + let (output, output_writer) = io::pipe().context("failed to create output pipe")?; + let stderr = output_writer + .try_clone() + .context("failed to share output pipe")?; + let (cancel, cancel_writer) = io::pipe().context("failed to create cancel pipe")?; + let (exited, exited_writer) = io::pipe().context("failed to create exit pipe")?; + Ok(Self { + output, + output_writer, + stderr, + cancel, + cancel_writer, + exited, + exited_writer, + }) + } +} + +impl TerminalHandle { + pub(super) fn spawn_pipe( + id: TerminalId, + order: u64, + request: CreateTerminal, + replay_capacity: usize, + ) -> Result { + let Pipes { + output, + output_writer, + stderr, + cancel, + cancel_writer, + exited, + exited_writer, + } = Pipes::new()?; + let mut command = Command::new(&request.program); + command + .args(&request.args) + .current_dir(&request.cwd) + .envs(&request.env) + .stdin(Stdio::null()) + .stdout(output_writer) + .stderr(stderr) + .process_group(0); + // SAFETY: only async-signal-safe calls run between fork and exec. + unsafe { + command.pre_exec(reset_signals); + } + let child = { + let _guard = SPAWN_LOCK.write().unwrap_or_else(PoisonError::into_inner); + command + .spawn() + .with_context(|| format!("failed to spawn {}", request.program))? + }; + // Close the daemon's copies of the write end, or EOF never arrives. + drop(command); + let process = Arc::new(Process { + pid: child.id(), + reaped: Mutex::new(false), + }); + // Dropping `Child` neither waits nor kills; `Process` reaps it. + drop(child); + + let info = new_info(id, ProcessMode::Pipe, Some(process.pid), &request); + // Nothing is ever written to a pipe process; a disconnected queue + // turns any stray write into an error. + let (writes, _) = bounded(1); + let actor = match start_actor( + id, + ProcessMode::Pipe, + ActorMaster { pty: None }, + writes, + &info, + replay_capacity, + (0, 0), + ) { + Ok(actor) => actor, + Err(error) => { + process.kill(); + process.reap(); + return Err(error); + } + }; + + let mut joins = vec![actor.thread]; + let reader_actor_tx = actor.actor_tx.clone(); + let started = thread::Builder::new() + .name(format!("opencode-pty-reader-{id}")) + .spawn(move || read_output(output, cancel, exited, reader_actor_tx)) + .context("failed to spawn terminal reader") + .and_then(|reader| { + joins.push(reader); + let wait_actor_tx = actor.actor_tx.clone(); + let pid = process.pid; + // Spawned last: the waiter must be joined before `reap`. + thread::Builder::new() + .name(format!("opencode-pty-wait-{id}")) + .spawn(move || { + let message = match wait_exit(pid) { + Ok(Exit::Code(code)) => ActorMessage::ChildExited(Ok(Some(code))), + Ok(Exit::Signal(name)) => ActorMessage::ChildKilled(name), + Err(error) => ActorMessage::ChildExited(Err(error.to_string())), + }; + let _ = wait_actor_tx.send(message); + // Starts the reader's drain deadline. + drop(exited_writer); + }) + .context("failed to spawn terminal child waiter") + }); + match started { + Ok(waiter) => joins.push(waiter), + Err(error) => { + // Stop what did start, kill the child before any join can wait + // on it, and reap only after every thread is gone. + drop(actor.shutdown_tx); + drop(cancel_writer); + process.kill(); + for join in joins { + let _ = join.join(); + } + process.reap(); + return Err(error); + } + } + + let reap = Arc::clone(&process); + Ok(Self { + order, + mode: ProcessMode::Pipe, + actor_tx: actor.actor_tx, + actor_done: actor.actor_done, + shutdown_tx: Mutex::new(Some(actor.shutdown_tx)), + killer: Mutex::new(Box::new(GroupKiller(Arc::clone(&process)))), + cancel: Mutex::new(Some(Box::new(cancel_writer))), + reap: Mutex::new(Some(Box::new(move || reap.reap()))), + info, + joins: Mutex::new(Some(joins)), + }) + } +} + +/// Runs in the child between fork and exec. The daemon blocks SIGTERM and +/// SIGINT on all its threads, and std keeps the signal mask across exec, so +/// clear it, and give every signal portable-pty also resets its default action. +fn reset_signals() -> io::Result<()> { + // SAFETY: sigemptyset, sigprocmask and signal are async-signal-safe. + unsafe { + // Dispositions first, so no inherited handler can run once unmasked. + for signal in [ + libc::SIGCHLD, + libc::SIGHUP, + libc::SIGINT, + libc::SIGQUIT, + libc::SIGTERM, + libc::SIGALRM, + ] { + libc::signal(signal, libc::SIG_DFL); + } + let mut empty: libc::sigset_t = std::mem::zeroed(); + if libc::sigemptyset(&mut empty) != 0 + || libc::sigprocmask(libc::SIG_SETMASK, &empty, std::ptr::null_mut()) != 0 + { + return Err(io::Error::last_os_error()); + } + } + Ok(()) +} + +/// A pipe child that stays a zombie after it exits, until `reap`. +#[derive(Debug)] +struct Process { + pid: u32, + reaped: Mutex, +} + +impl Process { + /// Kills the whole process group. Never signals after `reap`, when the + /// group ID may belong to someone else. + fn kill(&self) { + let reaped = self.reaped.lock().unwrap_or_else(PoisonError::into_inner); + if !*reaped && let Ok(pid) = i32::try_from(self.pid) { + // SAFETY: the unreaped child keeps its PID and group ID reserved. + // Signal the child directly too, in case it moved to another group. + unsafe { + libc::killpg(pid, libc::SIGKILL); + libc::kill(pid, libc::SIGKILL); + } + } + } + + /// Call only after the waiter has returned: kill the group first, so the + /// child has exited and this does not block. + fn reap(&self) { + let mut reaped = self.reaped.lock().unwrap_or_else(PoisonError::into_inner); + if *reaped { + return; + } + let Ok(pid) = i32::try_from(self.pid) else { + return; + }; + loop { + // SAFETY: waits only for this daemon's own child. + let result = unsafe { libc::waitpid(pid, std::ptr::null_mut(), 0) }; + if result != -1 || Errno::last() != Errno::EINTR { + break; + } + } + *reaped = true; + } +} + +#[derive(Debug)] +struct GroupKiller(Arc); + +impl ChildKiller for GroupKiller { + fn kill(&mut self) -> io::Result<()> { + self.0.kill(); + Ok(()) + } + + fn clone_killer(&self) -> Box { + Box::new(Self(Arc::clone(&self.0))) + } +} + +enum Exit { + Code(u32), + Signal(String), +} + +/// Waits for the child to exit without reaping it. +fn wait_exit(pid: u32) -> io::Result { + loop { + // SAFETY: siginfo_t is plain data, and all-zero is a valid value. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: `info` is a valid out pointer; WNOWAIT leaves the zombie. + let result = unsafe { + libc::waitid( + libc::P_PID, + pid as libc::id_t, + &mut info, + libc::WEXITED | libc::WNOWAIT, + ) + }; + if result == -1 { + let error = io::Error::last_os_error(); + if error.kind() == io::ErrorKind::Interrupted { + continue; + } + return Err(error); + } + // SAFETY: waitid filled `info` for a SIGCHLD event. + let status = unsafe { info.si_status() }; + return Ok(match info.si_code { + libc::CLD_EXITED => Exit::Code(status as u32), + _ => Exit::Signal(signal_name(status)), + }); + } +} + +fn signal_name(signal: i32) -> String { + let name = match signal { + libc::SIGHUP => "SIGHUP", + libc::SIGINT => "SIGINT", + libc::SIGQUIT => "SIGQUIT", + libc::SIGILL => "SIGILL", + libc::SIGTRAP => "SIGTRAP", + libc::SIGABRT => "SIGABRT", + libc::SIGBUS => "SIGBUS", + libc::SIGFPE => "SIGFPE", + libc::SIGKILL => "SIGKILL", + libc::SIGUSR1 => "SIGUSR1", + libc::SIGSEGV => "SIGSEGV", + libc::SIGUSR2 => "SIGUSR2", + libc::SIGPIPE => "SIGPIPE", + libc::SIGALRM => "SIGALRM", + libc::SIGTERM => "SIGTERM", + libc::SIGXCPU => "SIGXCPU", + libc::SIGXFSZ => "SIGXFSZ", + libc::SIGVTALRM => "SIGVTALRM", + libc::SIGPROF => "SIGPROF", + libc::SIGSYS => "SIGSYS", + _ => return format!("SIG{signal}"), + }; + name.to_string() +} + +/// Forwards output until EOF, until [`DRAIN_AFTER_EXIT`] after `exited` +/// closes, or until `cancel` closes. Once every writer has closed the output +/// pipe, what is left is finite, so it is read to the end regardless of the +/// deadline. +fn read_output( + output: PipeReader, + cancel: PipeReader, + exited: PipeReader, + actor: Sender, +) { + let mut output = output; + let mut buffer = [0_u8; 8192]; + let mut deadline = None::; + let mut writers_closed = false; + loop { + let timeout = match deadline { + Some(deadline) if !writers_closed => { + let left = deadline.saturating_duration_since(Instant::now()); + if left.is_zero() { + // The reader may have been stalled past the deadline; take + // one last look before giving up on a held-open pipe. + let mut fds = [PollFd::new(output.as_fd(), PollFlags::POLLIN)]; + let hung_up = poll(&mut fds, PollTimeout::ZERO).is_ok_and(|count| count > 0) + && fds[0] + .revents() + .is_some_and(|events| events.contains(PollFlags::POLLHUP)); + if !hung_up { + let _ = actor.send(ActorMessage::ReaderEof); + return; + } + writers_closed = true; + PollTimeout::NONE + } else { + // Round up: a sub-millisecond remainder must not become a busy 0. + let millis = left.as_nanos().div_ceil(1_000_000).min(i32::MAX as u128); + PollTimeout::try_from(millis).unwrap_or(PollTimeout::MAX) + } + } + _ => PollTimeout::NONE, + }; + let mut fds = [ + PollFd::new(output.as_fd(), PollFlags::POLLIN), + PollFd::new(cancel.as_fd(), PollFlags::POLLIN), + PollFd::new(exited.as_fd(), PollFlags::POLLIN), + ]; + // Once the exit is seen, stop polling its closed pipe. + let count = if deadline.is_some() { 2 } else { 3 }; + match poll(&mut fds[..count], timeout) { + Err(Errno::EINTR) | Ok(0) => continue, + Err(error) => { + let _ = actor.send(ActorMessage::ReaderFailed(error.to_string())); + return; + } + Ok(_) => {} + } + let ready = |fd: &PollFd| { + fd.revents().is_none_or(|events| { + events.intersects( + PollFlags::POLLIN + | PollFlags::POLLHUP + | PollFlags::POLLERR + | PollFlags::POLLNVAL, + ) + }) + }; + if ready(&fds[1]) { + return; + } + if count == 3 && ready(&fds[2]) { + deadline = Some(Instant::now() + DRAIN_AFTER_EXIT); + } + if fds[0] + .revents() + .is_some_and(|events| events.contains(PollFlags::POLLHUP)) + { + writers_closed = true; + } + if ready(&fds[0]) { + match output.read(&mut buffer) { + Ok(0) => { + let _ = actor.send(ActorMessage::ReaderEof); + return; + } + Ok(length) => { + if actor + .send(ActorMessage::Output(buffer[..length].to_vec())) + .is_err() + { + return; + } + } + Err(error) if error.kind() == io::ErrorKind::Interrupted => {} + Err(error) => { + let _ = actor.send(ActorMessage::ReaderFailed(error.to_string())); + return; + } + } + } + } +} diff --git a/packages/process-daemon/src/transport/unix.rs b/packages/process-daemon/src/transport/unix.rs index d66ead3da36d..5be4764267a6 100644 --- a/packages/process-daemon/src/transport/unix.rs +++ b/packages/process-daemon/src/transport/unix.rs @@ -391,6 +391,7 @@ mod tests { &mut server, &Response::Exited { exit_code: Some(0), + signal: None, final_offset: 123, }, ) @@ -401,6 +402,7 @@ mod tests { read_frame(&mut client).unwrap(), Response::Exited { exit_code: Some(0), + signal: None, final_offset: 123, } )); diff --git a/packages/process-daemon/src/transport/windows.rs b/packages/process-daemon/src/transport/windows.rs index edfdc3d92df6..cc9d04405d23 100644 --- a/packages/process-daemon/src/transport/windows.rs +++ b/packages/process-daemon/src/transport/windows.rs @@ -703,6 +703,7 @@ mod tests { &mut server, &Response::Exited { exit_code: Some(0), + signal: None, final_offset: 123, }, ) diff --git a/packages/process-daemon/tests/ownership.rs b/packages/process-daemon/tests/ownership.rs index 0cade5397a31..afadcfdfe1ed 100644 --- a/packages/process-daemon/tests/ownership.rs +++ b/packages/process-daemon/tests/ownership.rs @@ -110,6 +110,7 @@ impl Daemon { env: Default::default(), cols: 80, rows: 24, + mode: Default::default(), }) { Response::Created { terminal } => terminal, response => panic!("create: {response:?}"), @@ -261,6 +262,7 @@ fn subscription_delivers_final_exit_frame_before_closing() { Response::Exited { exit_code, final_offset, + .. } => { assert_eq!(exit_code, Some(0)); assert_eq!(final_offset, output.len() as u64); @@ -313,10 +315,10 @@ fn rename_is_advertised_and_served() { use opencode_pty::protocol::{SubscriptionEvent, read_subscription_event}; let mut daemon = Daemon::start(); - assert_eq!(daemon.registration.features, ["rename"]); + assert_eq!(daemon.registration.features, ["rename", "pipe"]); assert!(matches!( daemon.request(Request::Ping), - Response::Pong { features, .. } if features == ["rename"] + Response::Pong { features, .. } if features == ["rename", "pipe"] )); let (owner, response) = daemon.own(None); assert!(matches!(response, Response::Owned)); @@ -352,7 +354,7 @@ fn rename_is_advertised_and_served() { } } assert!( - matches!(daemon.request(Request::List), Response::Terminals { terminals } + matches!(daemon.request(Request::List { mode: None }), Response::Terminals { terminals } if terminals.len() == 1 && terminals[0].title == "renamed" && terminals[0].user_title.as_deref() == Some("renamed")) ); @@ -408,7 +410,7 @@ fn handoff_preserves_daemon_and_terminal_and_shutdown_overrides_it() { if instance_id == daemon.registration.instance_id && pid == daemon.child.id()) ); assert!( - matches!(daemon.request(Request::List), Response::Terminals { terminals } + matches!(daemon.request(Request::List { mode: None }), Response::Terminals { terminals } if terminals.len() == 1 && terminals[0].id == terminal.id && terminals[0].pid == terminal.pid) ); assert!(matches!( @@ -552,7 +554,8 @@ fn stop_signals_stop_terminals_and_clean_up() { assert_eq!(unsafe { libc::kill(probe.pid.unwrap() as i32, signal) }, 0); let deadline = Instant::now() + Duration::from_secs(3); loop { - let Response::Terminals { terminals } = daemon.request(Request::List) else { + let Response::Terminals { terminals } = daemon.request(Request::List { mode: None }) + else { panic!("list failed"); }; let exited = terminals.iter().any(|value| { @@ -575,3 +578,151 @@ fn stop_signals_stop_terminals_and_clean_up() { assert_terminal_stopped(&terminal); } } + +#[test] +fn pipe_processes_survive_handoff_and_stay_out_of_default_lists() { + use opencode_pty::protocol::{SubscriptionEvent, read_subscription_event}; + use opencode_pty::service::ProcessMode; + + let mut daemon = Daemon::start(); + let (owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + let terminal = daemon.terminal("exec cat"); + let Response::Created { terminal: process } = daemon.request(Request::Create { + program: "/bin/sh".into(), + args: vec![ + "-c".into(), + "echo before; sleep 1; echo after; exit 4".into(), + ], + cwd: daemon.directory.clone(), + title: String::new(), + group_id: "ownership".into(), + env: Default::default(), + cols: 0, + rows: 0, + mode: ProcessMode::Pipe, + }) else { + panic!("create pipe"); + }; + assert_eq!(process.mode, ProcessMode::Pipe); + let ids = |mode| match daemon.request(Request::List { mode }) { + Response::Terminals { terminals } => terminals + .into_iter() + .map(|terminal| terminal.id) + .collect::>(), + response => panic!("list: {response:?}"), + }; + assert_eq!(ids(None), vec![terminal.id]); + assert_eq!(ids(Some(ProcessMode::Pty)), vec![terminal.id]); + assert_eq!(ids(Some(ProcessMode::Pipe)), vec![process.id]); + assert!(matches!( + daemon.request(Request::Snapshot { id: process.id }), + Response::Error { message } if message.contains("pipe process") + )); + + // Restart handoff: the old owner goes away while the process runs. + let mut old = owner; + let Response::Handoff { ticket, .. } = daemon.send(&mut old, Request::PrepareHandoff) else { + panic!("handoff"); + }; + let (successor, response) = daemon.own(Some(ticket)); + assert!(matches!(response, Response::Owned)); + drop(old); + + let mut subscription = daemon.connect(); + let Response::Attached { replay_base64, .. } = daemon.send( + &mut subscription, + Request::Subscribe { + id: process.id, + offset: 0, + attachment_id: "pipe-observer".into(), + role: AttachmentRole::Observer, + takeover: false, + }, + ) else { + panic!("subscribe"); + }; + let mut output = base64::engine::general_purpose::STANDARD + .decode(replay_base64) + .unwrap(); + let exit = loop { + match read_subscription_event(&mut subscription).unwrap() { + SubscriptionEvent::Output { bytes, .. } => output.extend(bytes), + SubscriptionEvent::Response(response) => { + if let Response::Exited { exit_code, .. } = *response { + break exit_code; + } + } + } + }; + assert_eq!(exit, Some(4)); + assert_eq!(String::from_utf8(output).unwrap(), "before\nafter\n"); + assert!(matches!( + daemon.request(Request::Terminate { id: process.id }), + Response::Ok + )); + drop(subscription); + drop(successor); + daemon.wait(); + assert_terminal_stopped(&terminal); +} + +#[test] +fn pipe_processes_do_not_inherit_the_daemons_blocked_signals() { + use opencode_pty::service::{ProcessMode, TerminalLifecycle}; + + let mut daemon = Daemon::start(); + let (_owner, response) = daemon.own(None); + assert!(matches!(response, Response::Owned)); + for (signal, name) in [(libc::SIGTERM, "SIGTERM"), (libc::SIGINT, "SIGINT")] { + let Response::Created { terminal: process } = daemon.request(Request::Create { + program: "/bin/sh".into(), + args: vec!["-c".into(), "exec sleep 1000".into()], + cwd: daemon.directory.clone(), + title: String::new(), + group_id: "ownership".into(), + env: Default::default(), + cols: 0, + rows: 0, + mode: ProcessMode::Pipe, + }) else { + panic!("create pipe"); + }; + // SAFETY: the PID belongs to this test's pipe process. + assert_eq!( + unsafe { libc::kill(process.pid.unwrap() as i32, signal) }, + 0 + ); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let Response::Terminals { terminals } = daemon.request(Request::List { + mode: Some(ProcessMode::Pipe), + }) else { + panic!("list failed"); + }; + let lifecycle = terminals + .into_iter() + .find(|value| value.id == process.id) + .unwrap() + .lifecycle; + if lifecycle != TerminalLifecycle::Running { + assert_eq!( + lifecycle, + TerminalLifecycle::Exited { + exit_code: None, + signal: Some(name.into()), + } + ); + break; + } + assert!(Instant::now() < deadline, "pipe process ignored {name}"); + thread::sleep(Duration::from_millis(10)); + } + assert!(matches!( + daemon.request(Request::Terminate { id: process.id }), + Response::Ok + )); + } + assert!(matches!(daemon.request(Request::Shutdown), Response::Ok)); + daemon.wait(); +} diff --git a/packages/process-daemon/tests/pipe.rs b/packages/process-daemon/tests/pipe.rs new file mode 100644 index 000000000000..78fc1e986a43 --- /dev/null +++ b/packages/process-daemon/tests/pipe.rs @@ -0,0 +1,302 @@ +#![cfg(unix)] + +use std::env; +use std::time::{Duration, Instant}; + +use opencode_pty::protocol::AttachmentRole; +use opencode_pty::service::{ + CreateTerminal, ProcessMode, StreamEvent, TerminalId, TerminalLifecycle, TerminalService, +}; + +fn pipe(script: &str) -> CreateTerminal { + CreateTerminal { + program: "/bin/sh".to_string(), + args: vec!["-c".to_string(), script.to_string()], + cwd: env::current_dir().expect("cwd"), + title: String::new(), + group_id: "pipe".to_string(), + env: std::collections::HashMap::new(), + cols: 0, + rows: 0, + mode: ProcessMode::Pipe, + } +} + +struct Finished { + output: String, + exit_code: Option, + signal: Option, + final_offset: u64, + elapsed: Duration, +} + +/// Collects output from offset 0 until the exit event. +fn run(service: &TerminalService, id: TerminalId) -> Finished { + let started = Instant::now(); + let attachment = service + .attach(id, 0, "test".into(), AttachmentRole::Observer, false) + .unwrap(); + let mut output = attachment.replay.bytes.clone(); + loop { + match attachment + .events + .recv_timeout(Duration::from_secs(10)) + .expect("pipe process did not exit") + { + StreamEvent::Output { bytes, .. } => output.extend(bytes), + StreamEvent::Exited { + exit_code, + signal, + final_offset, + } => { + return Finished { + output: String::from_utf8(output).unwrap(), + exit_code, + signal, + final_offset, + elapsed: started.elapsed(), + }; + } + _ => {} + } + } +} + +fn alive(pid: i32) -> bool { + // SAFETY: signal zero only checks whether the PID exists. + unsafe { libc::kill(pid, 0) == 0 } +} + +fn wait_dead(pid: i32) { + let deadline = Instant::now() + Duration::from_secs(5); + while alive(pid) { + assert!(Instant::now() < deadline, "process {pid} is still running"); + std::thread::sleep(Duration::from_millis(10)); + } +} + +#[test] +fn reports_ordered_output_and_exit_code_without_a_terminal() { + let service = TerminalService::default(); + let info = service + .create(pipe( + "for i in 1 2 3; do echo out$i; echo err$i >&2; done + [ -t 1 ] && echo tty || echo notty + cat; echo stdin-closed + exit 7", + )) + .unwrap(); + assert_eq!(info.mode, ProcessMode::Pipe); + assert_eq!(info.title, "sh"); + let finished = run(&service, info.id); + assert_eq!( + finished.output, + "out1\nerr1\nout2\nerr2\nout3\nerr3\nnotty\nstdin-closed\n" + ); + assert_eq!(finished.exit_code, Some(7)); + assert_eq!(finished.signal, None); + assert_eq!(finished.final_offset, finished.output.len() as u64); + let listed = service.list_mode(ProcessMode::Pipe).unwrap(); + assert_eq!( + listed[0].lifecycle, + TerminalLifecycle::Exited { + exit_code: Some(7), + signal: None + } + ); + service.terminate(info.id).unwrap(); +} + +#[test] +fn sigpipe_is_not_inherited_as_ignored() { + // Rust ignores SIGPIPE in the daemon. A child that inherited that could + // never die from it, so `producer | head` pipelines would misbehave. + let service = TerminalService::default(); + let info = service + .create(pipe("kill -PIPE $$; echo survived")) + .unwrap(); + let finished = run(&service, info.id); + assert_eq!(finished.output, ""); + assert_eq!(finished.signal.as_deref(), Some("SIGPIPE")); + service.terminate(info.id).unwrap(); +} + +#[test] +fn reports_the_signal_that_ended_the_process() { + let service = TerminalService::default(); + let info = service.create(pipe("kill -TERM $$")).unwrap(); + let finished = run(&service, info.id); + assert_eq!(finished.exit_code, None); + assert_eq!(finished.signal.as_deref(), Some("SIGTERM")); + service.terminate(info.id).unwrap(); +} + +#[test] +fn fast_exit_without_output_is_reported() { + let service = TerminalService::default(); + let info = service.create(pipe("exit 3")).unwrap(); + let finished = run(&service, info.id); + assert_eq!(finished.output, ""); + assert_eq!(finished.exit_code, Some(3)); + assert_eq!(finished.final_offset, 0); + service.terminate(info.id).unwrap(); +} + +#[test] +fn a_descendant_holding_the_pipe_delays_the_exit_by_at_most_the_drain() { + let service = TerminalService::default(); + let info = service + .create(pipe("sleep 30 & echo $!; echo started")) + .unwrap(); + let finished = run(&service, info.id); + assert!( + finished.output.ends_with("started\n"), + "{}", + finished.output + ); + assert_eq!(finished.exit_code, Some(0)); + assert!( + finished.elapsed < Duration::from_secs(4), + "{:?}", + finished.elapsed + ); + let sleeper: i32 = finished.output.lines().next().unwrap().parse().unwrap(); + assert!(alive(sleeper)); + // Terminate kills the rest of the process group. + service.terminate(info.id).unwrap(); + wait_dead(sleeper); +} + +/// Kills a test descendant even when an assertion fails first. +struct KillOnDrop(i32); + +impl Drop for KillOnDrop { + fn drop(&mut self) { + // SAFETY: the PID belongs to this test's own descendant. + unsafe { libc::kill(self.0, libc::SIGKILL) }; + } +} + +#[test] +fn terminate_does_not_wait_for_a_descendant_that_left_the_group() { + let service = TerminalService::default(); + let info = service + .create(pipe( + "perl -e 'use POSIX; POSIX::setsid(); $| = 1; print \"$$\\n\"; sleep 30' & wait", + )) + .unwrap(); + let attachment = service + .attach(info.id, 0, "test".into(), AttachmentRole::Observer, false) + .unwrap(); + let mut output = attachment.replay.bytes.clone(); + while !output.ends_with(b"\n") { + if let StreamEvent::Output { bytes, .. } = attachment + .events + .recv_timeout(Duration::from_secs(5)) + .unwrap() + { + output.extend(bytes); + } + } + let escaped = KillOnDrop(String::from_utf8(output).unwrap().trim().parse().unwrap()); + drop(attachment); + let started = Instant::now(); + service.terminate(info.id).unwrap(); + assert!(started.elapsed() < Duration::from_secs(2)); + // It left the group, so it survives until `escaped` drops. + assert!(alive(escaped.0)); +} + +#[test] +fn an_exited_process_is_kept_until_terminate() { + let service = TerminalService::default(); + let info = service.create(pipe("exit 0")).unwrap(); + run(&service, info.id); + let pid = info.pid.unwrap() as i32; + // The zombie keeps the PID and process group reserved. + assert!(alive(pid)); + assert_eq!(service.list_mode(ProcessMode::Pipe).unwrap().len(), 1); + service.terminate(info.id).unwrap(); + assert!(!alive(pid)); + assert!(service.list_mode(ProcessMode::Pipe).unwrap().is_empty()); +} + +#[test] +fn terminate_kills_the_process_group() { + let service = TerminalService::default(); + let info = service.create(pipe("sleep 30 & echo $!; wait")).unwrap(); + let attachment = service + .attach(info.id, 0, "test".into(), AttachmentRole::Observer, false) + .unwrap(); + let mut output = attachment.replay.bytes.clone(); + while !output.ends_with(b"\n") { + if let StreamEvent::Output { bytes, .. } = attachment + .events + .recv_timeout(Duration::from_secs(5)) + .unwrap() + { + output.extend(bytes); + } + } + let child: i32 = String::from_utf8(output).unwrap().trim().parse().unwrap(); + drop(attachment); + service.terminate(info.id).unwrap(); + wait_dead(child); + assert!(!alive(info.pid.unwrap() as i32)); +} + +#[test] +fn terminal_requests_are_rejected_and_lists_are_separate() { + let service = TerminalService::default(); + let pty = service + .create(CreateTerminal { + cols: 80, + rows: 24, + mode: ProcessMode::Pty, + ..pipe("exec cat") + }) + .unwrap(); + let pipe = service.create(pipe("exec sleep 30")).unwrap(); + let id = pipe.id; + for result in [ + service.write(id, b"x".to_vec()), + service.resize(id, 80, 24), + service.control(id, "a".into(), 80, 24), + service.input(id, "a".into(), 80, 24, b"x".to_vec()), + service.rename(id, Some("x".into())), + service.snapshot(id).map(|_| ()), + service.read_rows(id, None).map(|_| ()), + ] { + let error = result.unwrap_err().to_string(); + assert!(error.contains("pipe process"), "{error}"); + } + assert!(service.replay(id, 0).is_ok()); + let listed = |mode| { + service + .list_mode(mode) + .unwrap() + .into_iter() + .map(|info| info.id) + .collect::>() + }; + assert_eq!(listed(ProcessMode::Pty), vec![pty.id]); + assert_eq!(listed(ProcessMode::Pipe), vec![pipe.id]); + assert_eq!(service.list().unwrap().len(), 1); + service.terminate(pty.id).unwrap(); + service.terminate(pipe.id).unwrap(); +} + +#[test] +fn pipe_processes_need_no_terminal_size() { + let service = TerminalService::default(); + assert!( + service + .create(CreateTerminal { + mode: ProcessMode::Pty, + ..pipe("true") + }) + .is_err() + ); + let info = service.create(pipe("true")).unwrap(); + service.terminate(info.id).unwrap(); +} diff --git a/packages/process-daemon/tests/runtime.rs b/packages/process-daemon/tests/runtime.rs index 27eab0d70c25..06d975f65502 100644 --- a/packages/process-daemon/tests/runtime.rs +++ b/packages/process-daemon/tests/runtime.rs @@ -297,6 +297,7 @@ fn check_normal_exit(code: u32) { StreamEvent::Exited { exit_code, final_offset, + .. } => { assert_eq!(exit_code, Some(code)); assert_eq!(final_offset, offset); @@ -314,7 +315,8 @@ fn check_normal_exit(code: u32) { assert_eq!( snapshot.info.lifecycle, TerminalLifecycle::Exited { - exit_code: Some(code) + exit_code: Some(code), + signal: None, } ); assert_eq!(snapshot.info.output_tail, offset); diff --git a/packages/process-daemon/tests/service.rs b/packages/process-daemon/tests/service.rs index 7293c6ae6a94..2988f3952b44 100644 --- a/packages/process-daemon/tests/service.rs +++ b/packages/process-daemon/tests/service.rs @@ -69,6 +69,7 @@ fn command(script: &str) -> CreateTerminal { env: std::collections::HashMap::new(), cols: 80, rows: 24, + mode: Default::default(), } } diff --git a/packages/process-daemon/tests/support/resize_fixture.rs b/packages/process-daemon/tests/support/resize_fixture.rs index e09ef8ab36ce..70f4577d06e1 100644 --- a/packages/process-daemon/tests/support/resize_fixture.rs +++ b/packages/process-daemon/tests/support/resize_fixture.rs @@ -43,6 +43,7 @@ impl ResizeFixture { env: [(CHILD.into(), listener.local_addr().unwrap().to_string())].into(), cols: 80, rows: 24, + mode: Default::default(), }) .unwrap(); diff --git a/packages/process-daemon/tests/support/terminal_fixture.rs b/packages/process-daemon/tests/support/terminal_fixture.rs index cdea300d2988..47b2f294a5e7 100644 --- a/packages/process-daemon/tests/support/terminal_fixture.rs +++ b/packages/process-daemon/tests/support/terminal_fixture.rs @@ -62,6 +62,7 @@ impl Fixture { .into(), cols: 80, rows: 24, + mode: Default::default(), } } diff --git a/packages/process-daemon/tests/windows-daemon.rs b/packages/process-daemon/tests/windows-daemon.rs index 86cf44856499..ac3a2e6e0d99 100644 --- a/packages/process-daemon/tests/windows-daemon.rs +++ b/packages/process-daemon/tests/windows-daemon.rs @@ -93,6 +93,7 @@ impl Daemon { env, cols, rows, + mode, } = request; match self.request(Request::Create { program, @@ -103,6 +104,7 @@ impl Daemon { env, cols, rows, + mode, }) { Response::Created { terminal } => terminal, response => panic!("create: {response:?}"), @@ -479,6 +481,7 @@ fn natural_exit_delivers_contiguous_output_and_retains_final_state() { Response::Exited { exit_code, final_offset, + .. } => { assert_eq!(final_offset, offset); break exit_code; @@ -504,7 +507,13 @@ fn natural_exit_delivers_contiguous_output_and_retains_final_state() { assert_eq!(exit_code, Some(actual_exit)); let (info, text) = daemon.snapshot(terminal.id); - assert_eq!(info.lifecycle, TerminalLifecycle::Exited { exit_code }); + assert_eq!( + info.lifecycle, + TerminalLifecycle::Exited { + exit_code, + signal: None + } + ); assert_eq!(info.output_tail, offset); assert!(text.contains("FINAL_DAEMON_MARKER"), "{text:?}"); let Response::Rows { @@ -557,7 +566,7 @@ fn natural_exit_delivers_contiguous_output_and_retains_final_state() { Response::Ok )); assert!( - matches!(daemon.request(Request::List), Response::Terminals { terminals } if terminals.is_empty()) + matches!(daemon.request(Request::List { mode: None }), Response::Terminals { terminals } if terminals.is_empty()) ); assert!(matches!(daemon.request(Request::Shutdown), Response::Ok)); drop(owner); @@ -648,7 +657,7 @@ fn observer_disconnect_and_control_input_preserve_independent_terminals() { assert!(!first_text.contains("SECOND_ONLY_MARKER")); assert!(!second_text.contains("FIRST_ONLY_MARKER")); assert!( - matches!(daemon.request(Request::List), Response::Terminals { terminals } if terminals.len() == 2) + matches!(daemon.request(Request::List { mode: None }), Response::Terminals { terminals } if terminals.len() == 2) ); assert!( matches!(daemon.request(Request::Ping), Response::Pong { instance_id, pid, protocol: 7, .. } @@ -662,7 +671,7 @@ fn observer_disconnect_and_control_input_preserve_independent_terminals() { )); drop(first_control); assert!( - matches!(daemon.request(Request::List), Response::Terminals { terminals } + matches!(daemon.request(Request::List { mode: None }), Response::Terminals { terminals } if terminals.len() == 1 && terminals[0].id == second.id) ); let input = b"after-first-exit";