From e2625605d5edfe8d5e5163d3d41599892ddec2ce Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Sun, 19 Oct 2025 11:40:30 -0700 Subject: [PATCH] feat: experimental `codex stdio-to-uds` subcommand --- codex-rs/Cargo.lock | 11 ++++ codex-rs/Cargo.toml | 2 + codex-rs/cli/Cargo.toml | 1 + codex-rs/cli/src/main.rs | 16 +++++ codex-rs/stdio-to-uds/Cargo.toml | 23 +++++++ codex-rs/stdio-to-uds/src/lib.rs | 60 +++++++++++++++++ codex-rs/stdio-to-uds/src/main.rs | 19 ++++++ codex-rs/stdio-to-uds/tests/stdio_to_uds.rs | 73 +++++++++++++++++++++ 8 files changed, 205 insertions(+) create mode 100644 codex-rs/stdio-to-uds/Cargo.toml create mode 100644 codex-rs/stdio-to-uds/src/lib.rs create mode 100644 codex-rs/stdio-to-uds/src/main.rs create mode 100644 codex-rs/stdio-to-uds/tests/stdio_to_uds.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9d18361cf8..706315f9a5 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -968,6 +968,7 @@ dependencies = [ "codex-protocol-ts", "codex-responses-api-proxy", "codex-rmcp-client", + "codex-stdio-to-uds", "codex-tui", "ctor 0.5.0", "owo-colors", @@ -1397,6 +1398,16 @@ dependencies = [ "webbrowser", ] +[[package]] +name = "codex-stdio-to-uds" +version = "0.0.0" +dependencies = [ + "anyhow", + "assert_cmd", + "pretty_assertions", + "tempfile", +] + [[package]] name = "codex-tui" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 64eebb62a1..93067af3a9 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -29,6 +29,7 @@ members = [ "protocol-ts", "rmcp-client", "responses-api-proxy", + "stdio-to-uds", "otel", "tui", "git-apply", @@ -73,6 +74,7 @@ codex-protocol = { path = "protocol" } codex-protocol-ts = { path = "protocol-ts" } codex-responses-api-proxy = { path = "responses-api-proxy" } codex-rmcp-client = { path = "rmcp-client" } +codex-stdio-to-uds = { path = "stdio-to-uds" } codex-tui = { path = "tui" } codex-utils-json-to-toml = { path = "utils/json-to-toml" } codex-utils-readiness = { path = "utils/readiness" } diff --git a/codex-rs/cli/Cargo.toml b/codex-rs/cli/Cargo.toml index 453ab80771..f5108a8d10 100644 --- a/codex-rs/cli/Cargo.toml +++ b/codex-rs/cli/Cargo.toml @@ -32,6 +32,7 @@ codex-app-server-protocol = { workspace = true } codex-protocol-ts = { workspace = true } codex-responses-api-proxy = { workspace = true } codex-tui = { workspace = true } +codex-stdio-to-uds = { workspace = true } codex-rmcp-client = { workspace = true } codex-cloud-tasks = { path = "../cloud-tasks" } ctor = { workspace = true } diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index e01bd259a5..19bd9e80d5 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -104,6 +104,10 @@ enum Subcommand { #[clap(hide = true)] ResponsesApiProxy(ResponsesApiProxyArgs), + /// Internal: relay stdio to a Unix domain socket. + #[clap(hide = true, name = "stdio-to-uds")] + StdioToUds(StdioToUdsCommand), + /// Inspect feature flags. Features(FeaturesCli), } @@ -205,6 +209,13 @@ struct GenerateTsCommand { prettier: Option, } +#[derive(Debug, Parser)] +struct StdioToUdsCommand { + /// Path to the Unix domain socket to connect to. + #[arg(value_name = "SOCKET_PATH")] + socket_path: PathBuf, +} + fn format_exit_messages(exit_info: AppExitInfo, color_enabled: bool) -> Vec { let AppExitInfo { token_usage, @@ -462,6 +473,11 @@ async fn cli_main(codex_linux_sandbox_exe: Option) -> anyhow::Result<() tokio::task::spawn_blocking(move || codex_responses_api_proxy::run_main(args)) .await??; } + Some(Subcommand::StdioToUds(cmd)) => { + let socket_path = cmd.socket_path; + tokio::task::spawn_blocking(move || codex_stdio_to_uds::run(socket_path.as_path())) + .await??; + } Some(Subcommand::GenerateTs(gen_cli)) => { codex_protocol_ts::generate_ts(&gen_cli.out_dir, gen_cli.prettier.as_deref())?; } diff --git a/codex-rs/stdio-to-uds/Cargo.toml b/codex-rs/stdio-to-uds/Cargo.toml new file mode 100644 index 0000000000..35b4d1987c --- /dev/null +++ b/codex-rs/stdio-to-uds/Cargo.toml @@ -0,0 +1,23 @@ +[package] +edition = "2024" +name = "codex-stdio-to-uds" +version = { workspace = true } + +[[bin]] +name = "codex-stdio-to-uds" +path = "src/main.rs" + +[lib] +name = "codex_stdio_to_uds" +path = "src/lib.rs" + +[lints] +workspace = true + +[dependencies] +anyhow = { workspace = true } + +[dev-dependencies] +assert_cmd = { workspace = true } +pretty_assertions = { workspace = true } +tempfile = { workspace = true } diff --git a/codex-rs/stdio-to-uds/src/lib.rs b/codex-rs/stdio-to-uds/src/lib.rs new file mode 100644 index 0000000000..8be5061a63 --- /dev/null +++ b/codex-rs/stdio-to-uds/src/lib.rs @@ -0,0 +1,60 @@ +#![deny(clippy::print_stdout)] + +use std::io; +use std::io::Read; +use std::io::Write; +use std::net::Shutdown; +use std::path::Path; +use std::thread; + +use anyhow::Context; +use anyhow::anyhow; + +#[cfg(unix)] +use std::os::unix::net::UnixStream; + +#[cfg(windows)] +use std::os::windows::net::UnixStream; + +/// Connects to the Unix Domain Socket at `socket_path` and relays data between +/// standard input/output and the socket. +pub fn run(socket_path: &Path) -> anyhow::Result<()> { + let stream = UnixStream::connect(socket_path) + .with_context(|| format!("failed to connect to socket at {}", socket_path.display()))?; + + let stdin = io::stdin(); + let stdout = io::stdout(); + + relay(stream, stdin, stdout) +} + +/// Relays data between the socket `stream`, the given input, and output +/// handles. Tests use this helper to inject custom IO streams. +pub fn relay(mut stream: UnixStream, mut input: R, output: W) -> anyhow::Result<()> +where + R: Read, + W: Write + Send + 'static, +{ + let mut reader = stream + .try_clone() + .context("failed to clone socket for reading")?; + + let stdout_thread = thread::spawn(move || -> io::Result<()> { + let mut output = output; + io::copy(&mut reader, &mut output)?; + output.flush()?; + Ok(()) + }); + + io::copy(&mut input, &mut stream).context("failed to copy data from input to socket")?; + stream + .shutdown(Shutdown::Write) + .context("failed to shutdown socket writer")?; + + let stdout_result = stdout_thread + .join() + .map_err(|_| anyhow!("thread panicked while copying socket data to output"))?; + stdout_result.context("failed to copy data from socket to output")?; + + Ok(()) +} diff --git a/codex-rs/stdio-to-uds/src/main.rs b/codex-rs/stdio-to-uds/src/main.rs new file mode 100644 index 0000000000..69987c31be --- /dev/null +++ b/codex-rs/stdio-to-uds/src/main.rs @@ -0,0 +1,19 @@ +use std::env; +use std::path::PathBuf; +use std::process; + +fn main() -> anyhow::Result<()> { + let mut args = env::args_os().skip(1); + let Some(socket_path) = args.next() else { + eprintln!("Usage: codex-stdio-to-uds "); + process::exit(1); + }; + + if args.next().is_some() { + eprintln!("Expected exactly one argument: "); + process::exit(1); + } + + let socket_path = PathBuf::from(socket_path); + codex_stdio_to_uds::run(&socket_path) +} diff --git a/codex-rs/stdio-to-uds/tests/stdio_to_uds.rs b/codex-rs/stdio-to-uds/tests/stdio_to_uds.rs new file mode 100644 index 0000000000..2b496e7c0c --- /dev/null +++ b/codex-rs/stdio-to-uds/tests/stdio_to_uds.rs @@ -0,0 +1,73 @@ +use std::fs; +use std::io::ErrorKind; +use std::io::Read; +use std::io::Write; +use std::sync::mpsc; +use std::thread; +use std::time::Duration; + +use anyhow::Context; +use assert_cmd::Command; +use pretty_assertions::assert_eq; + +#[cfg(unix)] +use std::os::unix::net::UnixListener; + +#[cfg(windows)] +use std::os::windows::net::UnixListener; + +#[test] +fn pipes_stdin_and_stdout_through_socket() -> anyhow::Result<()> { + let workspace_dir = std::env::current_dir().context("failed to read current dir")?; + let tmp_base = workspace_dir.join("target/stdio-to-uds-tests"); + fs::create_dir_all(&tmp_base).context("failed to create test temp base directory")?; + let dir = + tempfile::tempdir_in(&tmp_base).context("failed to create temp dir inside workspace")?; + let socket_path = dir.path().join("socket"); + let listener = match UnixListener::bind(&socket_path) { + Ok(listener) => listener, + Err(err) if err.kind() == ErrorKind::PermissionDenied => { + eprintln!("skipping test: failed to bind unix socket: {err}"); + return Ok(()); + } + Err(err) => { + return Err(err).context("failed to bind test unix socket"); + } + }; + + let (tx, rx) = mpsc::channel(); + let server_thread = thread::spawn(move || -> anyhow::Result<()> { + let (mut connection, _) = listener + .accept() + .context("failed to accept test connection")?; + let mut received = Vec::new(); + connection + .read_to_end(&mut received) + .context("failed to read data from client")?; + tx.send(received) + .map_err(|_| anyhow::anyhow!("failed to send received bytes to test thread"))?; + connection + .write_all(b"response") + .context("failed to write response to client")?; + Ok(()) + }); + + Command::cargo_bin("codex-stdio-to-uds")? + .arg(&socket_path) + .write_stdin("request") + .assert() + .success() + .stdout("response"); + + let received = rx + .recv_timeout(Duration::from_secs(1)) + .context("server did not receive data in time")?; + assert_eq!(received, b"request"); + + let server_result = server_thread + .join() + .map_err(|_| anyhow::anyhow!("server thread panicked"))?; + server_result.context("server failed")?; + + Ok(()) +}