diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a5fb378172..3c208054aa 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4380,6 +4380,22 @@ dependencies = [ "thiserror 2.0.18", ] +[[package]] +name = "codex-plugin-host" +version = "0.0.0" +dependencies = [ + "codex-utils-cargo-bin", + "codex-utils-process", + "codex-utils-pty", + "futures", + "libc", + "pretty_assertions", + "serde_json", + "tempfile", + "tokio", + "tokio-util", +] + [[package]] name = "codex-process-hardening" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 134c3f920b..830308f07e 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -30,6 +30,7 @@ members = [ "code-mode-host", "code-mode-protocol", "code-mode-runtime", + "plugin-host", "codex-home", "cloud-config", "cloud-tasks", diff --git a/codex-rs/plugin-host/BUILD.bazel b/codex-rs/plugin-host/BUILD.bazel new file mode 100644 index 0000000000..907fbc9cba --- /dev/null +++ b/codex-rs/plugin-host/BUILD.bazel @@ -0,0 +1,7 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "plugin-host", + crate_name = "codex_plugin_host", + test_only_binaries = ["codex-plugin-test-helper"], +) diff --git a/codex-rs/plugin-host/Cargo.toml b/codex-rs/plugin-host/Cargo.toml new file mode 100644 index 0000000000..31e56dffb2 --- /dev/null +++ b/codex-rs/plugin-host/Cargo.toml @@ -0,0 +1,39 @@ +[package] +name = "codex-plugin-host" +autotests = false +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +doctest = false +test = false + +[[bin]] +name = "codex-plugin-test-helper" +path = "tests/fixtures/plugin_process.rs" +test = false +bench = false +doc = false + +[[test]] +name = "process" +path = "tests/process.rs" + +[lints] +workspace = true + +[dependencies] +futures = { workspace = true, features = ["executor", "std"] } +tokio-util = { workspace = true, features = ["codec", "compat", "rt"] } +codex-utils-process = { workspace = true } +codex-utils-pty = { workspace = true } +libc = { workspace = true } +serde_json = { workspace = true } +tokio = { workspace = true, features = ["io-util", "macros", "process", "rt", "sync", "time"] } + +[dev-dependencies] +codex-utils-cargo-bin = { workspace = true } +pretty_assertions = { workspace = true } +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "net", "rt-multi-thread"] } diff --git a/codex-rs/plugin-host/src/framing.rs b/codex-rs/plugin-host/src/framing.rs new file mode 100644 index 0000000000..24e2e87e93 --- /dev/null +++ b/codex-rs/plugin-host/src/framing.rs @@ -0,0 +1,13 @@ +//! Shared wire-format configuration for both ends of the plugin pipes. + +use tokio_util::codec::LengthDelimitedCodec; + +pub const MAX_FRAME: usize = 256 * 1024; + +pub fn codec() -> LengthDelimitedCodec { + LengthDelimitedCodec::builder() + .little_endian() + .length_field_length(/*val*/ 4) + .max_frame_length(MAX_FRAME) + .new_codec() +} diff --git a/codex-rs/plugin-host/src/lib.rs b/codex-rs/plugin-host/src/lib.rs new file mode 100644 index 0000000000..677f24d03b --- /dev/null +++ b/codex-rs/plugin-host/src/lib.rs @@ -0,0 +1,11 @@ +//! Bounded IPC and lifetime management for the dedicated V8 plugin process. + +pub mod framing; +mod process; + +pub use process::PluginExit; +pub use process::PluginProcess; +pub use process::PluginProcessError; + +pub const HELPER_ARGUMENT: &str = "--codex-ui-plugin-process"; +pub const READY: &[u8] = b"codex-ui-plugin-process/2"; diff --git a/codex-rs/plugin-host/src/process.rs b/codex-rs/plugin-host/src/process.rs new file mode 100644 index 0000000000..faaa9c4495 --- /dev/null +++ b/codex-rs/plugin-host/src/process.rs @@ -0,0 +1,281 @@ +//! Owns the helper process group and preserves bounded diagnostics when it exits. + +use futures::Sink; +use futures::Stream; +use futures::StreamExt; +use std::fmt; +use std::io; +use std::path::Path; +use std::process::ExitStatus; +use std::process::Stdio; +use std::sync::Arc; +use tokio::io::AsyncReadExt; +use tokio::process::Child; +use tokio::process::ChildStdin; +use tokio::process::ChildStdout; +use tokio_util::bytes::Bytes; +use tokio_util::bytes::BytesMut; +use tokio_util::codec::FramedRead; +use tokio_util::codec::FramedWrite; +use tokio_util::codec::LengthDelimitedCodec; + +use crate::HELPER_ARGUMENT; +use crate::READY; +use crate::framing; + +const STDERR_TAIL_BYTES: usize = 16 * 1024; + +/// Process diagnostics. Raw stderr is untrusted and deliberately omitted from Debug. +pub struct PluginExit { + pub status: ExitStatus, + pub group_termination_error: Option, + /// At most the last 16 KiB, which may start mid-character or contain terminal escapes. + pub stderr_tail: Vec, + pub stderr_read_error: Option, +} + +impl fmt::Debug for PluginExit { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("PluginExit") + .field("status", &self.status) + .field("group_termination_error", &self.group_termination_error) + .field("stderr_tail_bytes", &self.stderr_tail.len()) + .field("stderr_read_error", &self.stderr_read_error) + .finish() + } +} + +/// A launch failure, or a readiness failure together with the outcome of closing the helper. +#[derive(Debug)] +pub enum PluginProcessError { + Launch(io::Error), + Handshake { + source: io::Error, + close_result: io::Result>, + }, +} + +impl From for PluginProcessError { + fn from(error: io::Error) -> Self { + Self::Launch(error) + } +} + +impl fmt::Display for PluginProcessError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Launch(error) => fmt::Display::fmt(error, f), + Self::Handshake { + source, + close_result: Ok(exit), + } => { + write!(f, "{source} ({})", exit.status)?; + if let Some(error) = &exit.group_termination_error { + write!(f, "; process-group cleanup failed: {error}")?; + } + Ok(()) + } + Self::Handshake { + source, + close_result: Err(error), + } => write!(f, "{source}; process cleanup failed: {error}"), + } + } +} + +impl std::error::Error for PluginProcessError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Launch(source) | Self::Handshake { source, .. } => Some(source), + } + } +} + +pub struct PluginProcess { + child: Child, + stderr: Option, Option)>>, + exit: Option>, + #[cfg(windows)] + job: codex_utils_pty::JobObject, + input: FramedWrite, + output: FramedRead, +} + +impl PluginProcess { + /// Borrow both pipes so callers can drive reads and writes concurrently. + pub fn pipes( + &mut self, + ) -> ( + &mut (impl Sink + Unpin), + &mut (impl Stream> + Unpin), + ) { + (&mut self.input, &mut self.output) + } + + /// Return a ready process, or reap it before returning a startup failure. + /// The caller owns the startup deadline. + /// Cancel through `stop` and await completion when teardown must finish before proceeding. + pub async fn spawn( + program: &Path, + stop: &tokio_util::sync::CancellationToken, + ) -> Result { + if !program.is_absolute() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "plugin helper must be absolute", + ) + .into()); + } + let program = std::fs::canonicalize(program)?; + let mut command = codex_utils_process::background_command(&program); + #[cfg(unix)] + { + use std::os::unix::process::CommandExt; + command.process_group(/*pgroup*/ 0); + } + command + .arg(HELPER_ARGUMENT) + .env_clear() + .current_dir( + program + .parent() + .ok_or_else(|| io::Error::other("invalid helper path"))?, + ) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut command = tokio::process::Command::from(command); + command.kill_on_drop(/*kill_on_drop*/ true); + let canceled = || io::Error::new(io::ErrorKind::Interrupted, "plugin startup canceled"); + if stop.is_cancelled() { + return Err(canceled().into()); + } + #[cfg(windows)] + let (mut child, job) = { + let job = codex_utils_pty::JobObject::create_without_breakaway()?; + let child = job.spawn_contained(&mut command)?; + (child, job) + }; + #[cfg(not(windows))] + let mut child = command.spawn()?; + let input = child + .stdin + .take() + .ok_or_else(|| io::Error::other("missing helper stdin"))?; + let output = child + .stdout + .take() + .ok_or_else(|| io::Error::other("missing helper stdout"))?; + let mut stderr = child + .stderr + .take() + .ok_or_else(|| io::Error::other("missing helper stderr"))?; + // Drain from startup so a full diagnostic pipe cannot block the handshake or exit. + let stderr = tokio::spawn(async move { + let mut tail = Vec::with_capacity(STDERR_TAIL_BYTES); + let mut buffer = [0; 4096]; + loop { + match stderr.read(&mut buffer).await { + Ok(0) => return (tail, None), + Ok(length) => { + let overflow = (tail.len() + length).saturating_sub(STDERR_TAIL_BYTES); + tail.drain(..overflow); + tail.extend_from_slice(&buffer[..length]); + } + Err(error) => return (tail, Some(error.kind())), + } + } + }); + let mut input = FramedWrite::new(input, framing::codec()); + // Keep only one accepted frame buffered when the child stops reading. + input.set_backpressure_boundary(/*boundary*/ 1); + let mut process = Self { + child, + stderr: Some(stderr), + exit: None, + #[cfg(windows)] + job, + input, + output: FramedRead::new(output, framing::codec()), + }; + let ready = async { + let bytes = process + .output + .next() + .await + .ok_or(io::ErrorKind::UnexpectedEof)??; + if bytes.as_ref() != READY { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "invalid plugin handshake", + )); + } + Ok(()) + }; + let result = tokio::select! { + biased; + _ = stop.cancelled() => Err(canceled()), + result = ready => result, + }; + if let Err(error) = result { + let close_result = process.close().await; + return Err(PluginProcessError::Handshake { + source: error, + close_result, + }); + } + Ok(process) + } + + /// Terminate the group, reap the child, and collect its exit status and diagnostic tail. + /// Group termination errors are retained in the report without losing the child's exit status. + /// The helper must not pass its stderr pipe to subprocesses; a surviving writer could make + /// diagnostic collection wait indefinitely after the helper exits. + /// Repeated calls after completion return the same report. A canceled call can be retried, + /// but the retry may lose a group-termination error from the canceled attempt. + pub async fn close(&mut self) -> io::Result> { + if let Some(exit) = &self.exit { + return Ok(exit.clone()); + } + let terminated = self.terminate_group(); + let reaped = self.child.kill().await; + reaped?; + let status = self.child.wait().await?; + // Borrow the task across await so canceling close does not detach the diagnostic owner. + let stderr = self + .stderr + .as_mut() + .ok_or_else(|| io::Error::other("missing stderr capture"))?; + let (stderr_tail, stderr_read_error) = stderr + .await + .unwrap_or_else(|_| (Vec::new(), Some(io::ErrorKind::Other))); + self.stderr.take(); + let exit = Arc::new(PluginExit { + status, + group_termination_error: terminated.err(), + stderr_tail, + stderr_read_error, + }); + self.exit = Some(exit.clone()); + Ok(exit) + } + + fn terminate_group(&self) -> io::Result<()> { + #[cfg(unix)] + if let Some(pid) = self.child.id() { + codex_utils_pty::process_group::kill_process_group(pid)?; + } + #[cfg(windows)] + self.job.terminate()?; + Ok(()) + } +} + +impl Drop for PluginProcess { + fn drop(&mut self) { + let _ = self.terminate_group(); + if let Some(stderr) = &self.stderr { + stderr.abort(); + } + } +} diff --git a/codex-rs/plugin-host/tests/fixtures/plugin_process.rs b/codex-rs/plugin-host/tests/fixtures/plugin_process.rs new file mode 100644 index 0000000000..25c7b1c399 --- /dev/null +++ b/codex-rs/plugin-host/tests/fixtures/plugin_process.rs @@ -0,0 +1,98 @@ +//! Test fixture for plugin process startup, framing, and lifecycle failures. + +use codex_plugin_host::framing; +use futures::SinkExt; +use futures::StreamExt; +use futures::executor::block_on; +use futures::io::AllowStdIo; +use std::io::Write; +use tokio_util::codec::FramedRead; +use tokio_util::codec::FramedWrite; +use tokio_util::compat::FuturesAsyncReadCompatExt; +use tokio_util::compat::FuturesAsyncWriteCompatExt; + +fn crash() -> std::io::Result<()> { + let mut stderr = std::io::stderr().lock(); + stderr.write_all(&vec![b'x'; 128 * 1024])?; + stderr.write_all(b"probe failed\n")?; + std::process::exit(/*code*/ 42); +} + +fn main() -> std::io::Result<()> { + let mut output = FramedWrite::new( + AllowStdIo::new(std::io::stdout().lock()).compat_write(), + framing::codec(), + ); + #[cfg(unix)] + if let Some(socket) = std::env::args_os() + .skip(1) + .find(|arg| std::path::Path::new(arg).is_absolute()) + { + let _connection = std::os::unix::net::UnixStream::connect(socket)?; + loop { + std::thread::park(); + } + } + if std::env::args().any(|arg| arg == "--descendant") { + block_on(output.send(b"descendant-ready".as_slice().into()))?; + loop { + std::thread::park(); + } + } + let program = std::env::current_exe()?; + if program.file_stem() == Some(std::ffi::OsStr::new("failing-plugin-probe")) { + return crash(); + } + #[cfg(unix)] + if program.file_stem() == Some(std::ffi::OsStr::new("unready-plugin-probe")) { + std::os::unix::net::UnixStream::connect(program.with_extension("sock"))? + .write_all(&std::process::id().to_le_bytes())?; + loop { + std::thread::park(); + } + } + let mut input = FramedRead::new( + AllowStdIo::new(std::io::stdin().lock()).compat(), + framing::codec(), + ); + block_on(output.send(codex_plugin_host::READY.into()))?; + #[cfg(unix)] + { + // SAFETY: getpgrp reads this process's group identifier. + let group = unsafe { libc::getpgrp() }; + block_on(output.send(serde_json::to_vec(&(std::process::id(), group))?.into()))?; + } + let mut descendants = Vec::new(); + while let Some(message) = block_on(input.next()) { + let message = message?; + #[cfg(unix)] + if message.starts_with(b"/") { + use std::os::unix::ffi::OsStrExt; + descendants.push( + codex_utils_process::background_command(&program) + .arg(std::ffi::OsStr::from_bytes(&message)) + .spawn()?, + ); + continue; + } + if message.as_ref() == b"crash" { + return crash(); + } else if message.as_ref() == b"spawn-descendant" { + descendants.push( + codex_utils_process::background_command(&program) + .arg("--descendant") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::inherit()) + .stderr(std::process::Stdio::null()) + .spawn()?, + ); + } else { + block_on(output.send(message.freeze()))?; + } + } + for mut descendant in descendants { + descendant.kill()?; + descendant.wait()?; + } + Ok(()) +} diff --git a/codex-rs/plugin-host/tests/process.rs b/codex-rs/plugin-host/tests/process.rs new file mode 100644 index 0000000000..5ab1479146 --- /dev/null +++ b/codex-rs/plugin-host/tests/process.rs @@ -0,0 +1,182 @@ +use codex_plugin_host::PluginProcess; +use codex_plugin_host::PluginProcessError; +use futures::SinkExt; +use futures::StreamExt; +use pretty_assertions::assert_eq; +#[cfg(unix)] +use tokio::io::AsyncReadExt; +use tokio_util::sync::CancellationToken; + +// Sibling forks can retain a copied executable's writable descriptor and cause ETXTBSY. +static PROCESS_FIXTURES: tokio::sync::Semaphore = + tokio::sync::Semaphore::const_new(/*permits*/ 1); + +#[tokio::test] +async fn owns_a_separate_process_group_and_reaps_the_helper() +-> Result<(), Box> { + let _fixture = PROCESS_FIXTURES.acquire().await?; + let program = codex_utils_cargo_bin::cargo_bin("codex-plugin-test-helper")?; + let mut child = PluginProcess::spawn(&program, &CancellationToken::new()).await?; + let (input, output) = child.pipes(); + #[cfg(unix)] + { + let bytes = output.next().await.expect("expected frame")?; + let (pid, group): (u32, i32) = serde_json::from_slice(&bytes)?; + assert_eq!(group, pid as i32); + // SAFETY: getpgrp reads the test process's group identifier. + assert_ne!(group, unsafe { libc::getpgrp() }); + } + input.feed(b"hello".to_vec().into()).await?; + input.flush().await?; + assert_eq!( + output.next().await.expect("expected frame")?.as_ref(), + b"hello" + ); + input.feed(b"spawn-descendant".to_vec().into()).await?; + input.flush().await?; + assert_eq!( + output.next().await.expect("expected frame")?.as_ref(), + b"descendant-ready" + ); + child.close().await?; + assert!(child.pipes().1.next().await.transpose()?.is_none()); + child.close().await?; + Ok(()) +} + +#[tokio::test] +async fn already_canceled_startup_does_not_launch_the_helper() +-> Result<(), Box> { + let _fixture = PROCESS_FIXTURES.acquire().await?; + let program = codex_utils_cargo_bin::cargo_bin("codex-plugin-test-helper")?; + let stop = CancellationToken::new(); + stop.cancel(); + let error = PluginProcess::spawn(&program, &stop) + .await + .err() + .expect("canceled startup succeeded"); + let PluginProcessError::Launch(source) = error else { + panic!("expected a launch error, got {error:?}"); + }; + assert_eq!(source.kind(), std::io::ErrorKind::Interrupted); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn dropping_the_helper_also_closes_descendant_pipes() -> Result<(), Box> +{ + let _fixture = PROCESS_FIXTURES.acquire().await?; + let directory = tempfile::tempdir()?; + let socket = directory.path().join("descendant.sock"); + let listener = tokio::net::UnixListener::bind(&socket)?; + + let program = codex_utils_cargo_bin::cargo_bin("codex-plugin-test-helper")?; + let mut child = PluginProcess::spawn(&program, &CancellationToken::new()).await?; + let (input, output) = child.pipes(); + let _group = output.next().await.expect("expected frame")?; + input + .send(socket.as_os_str().as_encoded_bytes().to_vec().into()) + .await?; + let (mut descendant, _) = listener.accept().await?; + drop(child); + assert_eq!(descendant.read(&mut [0]).await?, 0); + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn caller_can_cancel_a_stalled_handshake_and_reap_the_helper() +-> Result<(), Box> { + let _fixture = PROCESS_FIXTURES.acquire().await?; + let directory = tempfile::tempdir()?; + let program = directory.path().join("unready-plugin-probe"); + std::fs::copy( + codex_utils_cargo_bin::cargo_bin("codex-plugin-test-helper")?, + &program, + )?; + let listener = tokio::net::UnixListener::bind(program.with_extension("sock"))?; + let stop = CancellationToken::new(); + let spawn = PluginProcess::spawn(&program, &stop); + tokio::pin!(spawn); + let pid = tokio::select! { + _ = &mut spawn => panic!("unready helper completed startup"), + pid = async { + let (mut signal, _) = listener.accept().await?; + signal.read_i32_le().await + } => pid?, + }; + stop.cancel(); + let error = match spawn.await { + Ok(_) => panic!("canceled startup succeeded"), + Err(error) => error, + }; + let PluginProcessError::Handshake { + source, + close_result, + } = error + else { + panic!("expected a handshake error, got {error:?}"); + }; + assert_eq!(source.kind(), std::io::ErrorKind::Interrupted); + close_result?; + // SAFETY: getpgid only inspects the child recorded by the test probe. + assert_eq!(unsafe { libc::getpgid(pid) }, -1); + Ok(()) +} + +#[tokio::test] +async fn preserves_exit_status_and_bounded_stderr_from_startup_and_runtime_failures() +-> Result<(), Box> { + let _fixture = PROCESS_FIXTURES.acquire().await?; + let directory = tempfile::tempdir()?; + let failing = directory.path().join(format!( + "failing-plugin-probe{}", + std::env::consts::EXE_SUFFIX + )); + let program = codex_utils_cargo_bin::cargo_bin("codex-plugin-test-helper")?; + std::fs::copy(&program, &failing)?; + for executable in [&failing, &program] { + let exit = if executable == &failing { + let error = match PluginProcess::spawn(executable, &CancellationToken::new()).await { + Ok(_) => panic!("failing helper started"), + Err(error) => error, + }; + let PluginProcessError::Handshake { + source, + close_result, + } = error + else { + panic!("expected a handshake error, got {error:?}"); + }; + assert_eq!(source.kind(), std::io::ErrorKind::UnexpectedEof); + close_result? + } else { + let mut child = PluginProcess::spawn(executable, &CancellationToken::new()).await?; + let (input, output) = child.pipes(); + #[cfg(unix)] + let _group = output.next().await.expect("expected frame")?; + input.feed(b"crash".to_vec().into()).await?; + input.flush().await?; + assert!(output.next().await.transpose()?.is_none()); + let exit = child.close().await?; + assert!(std::sync::Arc::ptr_eq(&exit, &child.close().await?)); + exit + }; + let mut expected = vec![b'x'; 16 * 1024 - b"probe failed\n".len()]; + expected.extend_from_slice(b"probe failed\n"); + assert_eq!( + ( + exit.status.code(), + &exit.stderr_tail, + exit.group_termination_error + .as_ref() + .map(std::io::Error::kind), + exit.stderr_read_error + ), + (Some(42), &expected, None, None) + ); + assert!(!format!("{exit:?}").contains("probe failed")); + } + Ok(()) +} diff --git a/defs.bzl b/defs.bzl index 7fe51d40bc..6d284ccd61 100644 --- a/defs.bzl +++ b/defs.bzl @@ -193,6 +193,7 @@ def codex_rust_crate( build_script_data = [], compile_data = [], binary_compile_data_extra = {}, + test_only_binaries = [], lib_data_extra = [], rustc_flags_extra = [], binary_rustc_flags_extra = {}, @@ -238,6 +239,7 @@ def codex_rust_crate( compile_data: Non-Rust compile-time data for the library target. binary_compile_data_extra: Mapping from binary names to extra non-Rust compile-time data for those binary targets. + test_only_binaries: Binary names used only as test fixtures. lib_data_extra: Extra runtime data for the library target. binary_rustc_flags_extra: Mapping from binary names to extra rustc flags for those binary targets. @@ -432,7 +434,8 @@ def codex_rust_crate( # Isolate status inputs in build-commit-env's cheap action so its # output, and hence this compiler input, changes only with the commit. rustc_env_files = ["//bazel/build-info:build-commit-env"] if binary in binaries_with_build_commit else [], - srcs = native.glob(["src/**/*.rs"]), + srcs = native.glob(["src/**/*.rs"], exclude = [main], allow_empty = True) + [main], + testonly = binary in test_only_binaries, stamp = 0, version = crate_version, visibility = ["//visibility:public"],