diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e9811e7a57..2c22dde85b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4042,6 +4042,14 @@ dependencies = [ "uuid", ] +[[package]] +name = "codex-realtime-webrtc" +version = "0.0.0" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "codex-response-debug-context" version = "0.0.0" @@ -4800,6 +4808,18 @@ dependencies = [ "v8", ] +[[package]] +name = "codex-voice-host" +version = "0.0.0" +dependencies = [ + "anyhow", + "codex-process-hardening", + "codex-realtime-webrtc", + "codex-utils-cargo-bin", + "pretty_assertions", + "tokio", +] + [[package]] name = "codex-web-search-extension" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 315ddc120d..c6e21960f5 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -85,6 +85,8 @@ members = [ "network-proxy", "ollama", "process-hardening", + "realtime-webrtc", + "voice-host", "protocol", "prompts", "rollout", @@ -152,6 +154,7 @@ license = "Apache-2.0" [workspace.dependencies] # Internal +codex-realtime-webrtc = { path = "realtime-webrtc" } app_test_support = { path = "app-server/tests/common" } codex-analytics = { path = "analytics" } codex-agent-graph-store = { path = "agent-graph-store" } diff --git a/codex-rs/realtime-webrtc/BUILD.bazel b/codex-rs/realtime-webrtc/BUILD.bazel new file mode 100644 index 0000000000..d9cfeb6cfa --- /dev/null +++ b/codex-rs/realtime-webrtc/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "realtime-webrtc", + crate_name = "codex_realtime_webrtc", +) diff --git a/codex-rs/realtime-webrtc/Cargo.toml b/codex-rs/realtime-webrtc/Cargo.toml new file mode 100644 index 0000000000..218a5858d1 --- /dev/null +++ b/codex-rs/realtime-webrtc/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "codex-realtime-webrtc" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lib] +doctest = false +test = false + +[lints] +workspace = true + +[dependencies] +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } diff --git a/codex-rs/realtime-webrtc/src/lib.rs b/codex-rs/realtime-webrtc/src/lib.rs new file mode 100644 index 0000000000..ac68e9311b --- /dev/null +++ b/codex-rs/realtime-webrtc/src/lib.rs @@ -0,0 +1,7 @@ +mod protocol; + +pub use protocol::MAX_FRAME_BYTES; +pub use protocol::Message; +pub use protocol::decode_frame; +pub use protocol::encode_frame; +pub use protocol::read_message; diff --git a/codex-rs/realtime-webrtc/src/protocol.rs b/codex-rs/realtime-webrtc/src/protocol.rs new file mode 100644 index 0000000000..847529facb --- /dev/null +++ b/codex-rs/realtime-webrtc/src/protocol.rs @@ -0,0 +1,63 @@ +//! Bounded, same-build lifecycle protocol; this stage carries no audio or credentials. + +use std::io; +use std::io::Read; + +use serde::Deserialize; +use serde::Serialize; + +pub const MAX_FRAME_BYTES: usize = 256; + +#[derive(Debug, Deserialize, PartialEq, Serialize)] +#[serde( + tag = "type", + rename_all = "camelCase", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum Message { + Hello { protocol: u32, build_commit: String }, + Ready {}, + Close {}, + Closed {}, +} + +pub fn encode_frame(message: &Message) -> io::Result> { + let payload = serde_json::to_vec(message)?; + if payload.len() > MAX_FRAME_BYTES { + return Err(io::Error::other("voice frame exceeds limit")); + } + let mut frame = (payload.len() as u32).to_be_bytes().to_vec(); + frame.extend(payload); + Ok(frame) +} + +pub fn decode_frame(frame: &[u8]) -> io::Result> { + let [a, b, c, d, ..] = frame else { + return Ok(None); + }; + let length = u32::from_be_bytes([*a, *b, *c, *d]) as usize; + if length > MAX_FRAME_BYTES || frame.len() > length + 4 { + return Err(io::Error::other("invalid voice frame length")); + } + frame + .get(4..length + 4) + .map(serde_json::from_slice) + .transpose() + .map_err(|_| io::Error::other("invalid voice frame")) +} + +pub fn read_message(reader: &mut impl Read) -> io::Result> { + let mut header = [0; 4]; + match reader.read_exact(&mut header[..1]) { + Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => return Ok(None), + result => result?, + } + reader.read_exact(&mut header[1..])?; + decode_frame(&header)?; + let length = u32::from_be_bytes(header) as usize; + let mut frame = header.to_vec(); + frame.resize(length + 4, /*value*/ 0); + reader.read_exact(&mut frame[4..])?; + decode_frame(&frame) +} diff --git a/codex-rs/voice-host/BUILD.bazel b/codex-rs/voice-host/BUILD.bazel new file mode 100644 index 0000000000..fca93d945d --- /dev/null +++ b/codex-rs/voice-host/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "voice-host", + crate_name = "codex_voice_host", +) diff --git a/codex-rs/voice-host/Cargo.toml b/codex-rs/voice-host/Cargo.toml new file mode 100644 index 0000000000..e8b7248c51 --- /dev/null +++ b/codex-rs/voice-host/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "codex-voice-host" +version.workspace = true +edition.workspace = true +license.workspace = true + +[[bin]] +name = "codex-voice-host" +path = "src/main.rs" + +[lints] +workspace = true + +[dependencies] +codex-process-hardening = { workspace = true } +codex-realtime-webrtc = { workspace = true } + +[dev-dependencies] +anyhow = { workspace = true } +codex-utils-cargo-bin = { workspace = true } +pretty_assertions = { workspace = true } +tokio = { workspace = true, features = ["io-util", "macros", "process", "rt-multi-thread", "time"] } diff --git a/codex-rs/voice-host/README.md b/codex-rs/voice-host/README.md new file mode 100644 index 0000000000..75ff1501c4 --- /dev/null +++ b/codex-rs/voice-host/README.md @@ -0,0 +1,17 @@ +# Private voice helper foundation + +`codex-voice-host` establishes the inherited-pipe lifecycle for the proposed +bundled voice process. It does not open devices, load native plugins, negotiate +WebRTC, or enable voice in the TUI. The existing CLI is unchanged. + +Frames are a big-endian u32 length followed by at most 256 bytes of JSON. The +parent sends `hello` with protocol `1` and the helper's exact `buildCommit` before +receiving `ready`. It then sends `close` and receives `closed` before process exit. +Unknown fields, incompatible builds, invalid order and oversized frames fail +closed without echoing input. EOF exits even when the main worker cannot progress. + +Bazel stamps the binary with `STABLE_GIT_COMMIT`. Cargo builders must provide the +same variable; an unstamped source build reports `dev` via `--build-commit` and is +not a distributable build identity. The client/control crate has no native audio +dependencies. Installed-package resolution, native loading, media/privacy controls +and actual audio proof belong to the subsequent integration stages. diff --git a/codex-rs/voice-host/src/main.rs b/codex-rs/voice-host/src/main.rs new file mode 100644 index 0000000000..b4760859d0 --- /dev/null +++ b/codex-rs/voice-host/src/main.rs @@ -0,0 +1,75 @@ +//! Private helper lifecycle foundation. No devices, native plugins, or media are opened yet. + +use std::io; +use std::io::Write; +use std::sync::mpsc; +use std::time::Duration; + +use codex_realtime_webrtc::Message; +use codex_realtime_webrtc::encode_frame; +use codex_realtime_webrtc::read_message; + +const BUILD_COMMIT: &str = match option_env!("STABLE_GIT_COMMIT") { + Some(commit) => commit, + None => "dev", +}; + +fn main() { + codex_process_hardening::pre_main_hardening(); + let mut args = std::env::args_os().skip(/*n*/ 1); + match (args.next(), args.next()) { + (Some(arg), None) if arg == "--build-commit" => println!("{BUILD_COMMIT}"), + (None, None) => { + if run().is_err() { + std::process::exit(/*code*/ 1); + } + } + _ => std::process::exit(/*code*/ 2), + } +} + +fn run() -> io::Result<()> { + let (sender, receiver) = mpsc::sync_channel(/*bound*/ 1); + std::thread::Builder::new() + .name("voice-control".into()) + .spawn(move || { + let mut input = io::stdin().lock(); + loop { + match read_message(&mut input) { + Ok(Some(message)) => { + if sender.try_send(message).is_err() { + std::process::exit(/*code*/ 1); + } + } + Ok(None) => break, + Err(_) => std::process::exit(/*code*/ 1), + } + } + drop(sender); + // Independent of the main worker or a blocked stdout write, including after parent death. + std::thread::sleep(Duration::from_secs(/*secs*/ 2)); + std::process::exit(/*code*/ 1); + })?; + let Ok(hello) = receiver.recv() else { + return Ok(()); + }; + if hello + != (Message::Hello { + protocol: 1, + build_commit: BUILD_COMMIT.to_owned(), + }) + { + return Err(io::Error::other("incompatible voice helper")); + } + let mut output = io::stdout().lock(); + output.write_all(&encode_frame(&Message::Ready {})?)?; + output.flush()?; + match receiver.recv() { + Ok(Message::Close {}) => { + output.write_all(&encode_frame(&Message::Closed {})?)?; + output.flush() + } + Err(_) => Ok(()), + Ok(_) => Err(io::Error::other("invalid voice control sequence")), + } +} diff --git a/codex-rs/voice-host/tests/lifecycle.rs b/codex-rs/voice-host/tests/lifecycle.rs new file mode 100644 index 0000000000..9f6eeb04fb --- /dev/null +++ b/codex-rs/voice-host/tests/lifecycle.rs @@ -0,0 +1,123 @@ +//! Exercise the real helper through installed paths and bounded process I/O. + +use std::process::Stdio; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use codex_realtime_webrtc::Message; +use codex_realtime_webrtc::decode_frame; +use codex_realtime_webrtc::encode_frame; +use codex_utils_cargo_bin::cargo_bin; +use pretty_assertions::assert_eq; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::process::Child; +use tokio::process::Command; +use tokio::time::timeout; + +const DEADLINE: Duration = Duration::from_secs(/*secs*/ 10); + +fn spawn() -> Result { + Ok(Command::new(cargo_bin("codex-voice-host")?) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn()?) +} + +async fn build_commit() -> Result { + let output = timeout( + DEADLINE, + Command::new(cargo_bin("codex-voice-host")?) + .arg("--build-commit") + .kill_on_drop(true) + .output(), + ) + .await??; + assert!(output.status.success()); + Ok(String::from_utf8(output.stdout)?.trim().to_owned()) +} + +async fn handshake(child: &mut Child) -> Result<()> { + child + .stdin + .as_mut() + .context("stdin")? + .write_all(&encode_frame(&Message::Hello { + protocol: 1, + build_commit: build_commit().await?, + })?) + .await?; + let expected = encode_frame(&Message::Ready {})?; + let mut reply = vec![0; expected.len()]; + timeout( + DEADLINE, + child + .stdout + .as_mut() + .context("stdout")? + .read_exact(&mut reply), + ) + .await??; + assert_eq!(reply, expected); + Ok(()) +} + +#[tokio::test] +async fn closes_after_acknowledgement_and_on_parent_pipe_loss() -> Result<()> { + for explicit_close in [true, false] { + let mut child = spawn()?; + handshake(&mut child).await?; + if explicit_close { + child + .stdin + .as_mut() + .unwrap() + .write_all(&encode_frame(&Message::Close {})?) + .await?; + } + drop(child.stdin.take()); + let output = timeout(DEADLINE, child.wait_with_output()).await??; + assert!(output.status.success()); + assert_eq!( + output.stdout, + if explicit_close { + encode_frame(&Message::Closed {})? + } else { + vec![] + } + ); + assert!(output.stderr.is_empty()); + } + Ok(()) +} + +#[tokio::test] +async fn rejects_invalid_input_without_echoing_it() -> Result<()> { + assert!(decode_frame(b"\0\0\0\x16{\"type\":\"close\",\"x\":0}").is_err()); + let mut invalid_json = 22_u32.to_be_bytes().to_vec(); + invalid_json.extend_from_slice(b"sensitive-invalid-json"); + for frame in [ + u32::MAX.to_be_bytes().to_vec(), + vec![0, 0], + invalid_json, + encode_frame(&Message::Hello { + protocol: 1, + build_commit: "wrong-build".into(), + })?, + encode_frame(&Message::Hello { + protocol: 99, + build_commit: build_commit().await?, + })?, + encode_frame(&Message::Close {})?, + ] { + let mut child = spawn()?; + child.stdin.take().unwrap().write_all(&frame).await?; + let output = timeout(DEADLINE, child.wait_with_output()).await??; + assert!(!output.status.success()); + assert_eq!((output.stdout, output.stderr), (vec![], vec![])); + } + Ok(()) +}