Add a plugin process host with bounded IPC and lifecycle management (#57290)

## What changed - Add `codex-plugin-host` to launch a helper with a cleared environment, verify its readiness handshake, and expose framed stdin/stdout pipes with a 256 KiB frame limit and write backpressure. - Support startup cancellation and process-group cleanup, retaining exit status and the last 16 KiB of stderr while omitting raw diagnostics from `Debug` output. - Extend the Bazel crate macro to support test-only binaries and fixture entry points outside `src/`. ## Testing Add process integration tests for framed communication, startup cancellation, helper and descendant cleanup, repeated close calls, and bounded diagnostics from startup and runtime failures. GitOrigin-RevId: ed539e7899af06ee83a0d8c2d0ea8cec10f19afb
2026-10-28 18:58:46 +00:00 · 2026-10-26 22:11:01 +00:00
parent b46a3801e4
commit 16effdfc8c
10 changed files with 652 additions and 1 deletions
@@ -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(())
+}
@@ -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<dyn std::error::Error>> {
+    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<dyn std::error::Error>> {
+    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<dyn std::error::Error>>
+{
+    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<dyn std::error::Error>> {
+    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<dyn std::error::Error>> {
+    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(())
+}