Add the voice helper lifecycle foundation (#41897)

## What changed

- Add `codex-realtime-webrtc` with a length-prefixed, 256-byte JSON control protocol for `hello`, `ready`, `close`, and `closed` messages.
- Add a hardened `codex-voice-host` binary that validates protocol and build compatibility, handles orderly shutdown and parent pipe loss, and exposes its build commit.
- Keep this initial foundation limited to process lifecycle management; it does not open audio devices, load native plugins, negotiate WebRTC, or enable voice in the TUI.

## Testing

- Exercise the helper as a subprocess, covering handshake, explicit close, parent pipe loss, malformed frames, incompatible builds, and invalid message order.

GitOrigin-RevId: 6d8d7dd7ba5c4886879882d79314d235f6ab7cf4
This commit is contained in:
Benjamin Carlsson authored and copyberry committed 2026-08-31 19:08:43 +00:00
1 parent 65237aeca0
commit d60560f14e
11 files changed
+358

No files matched your search

+20
View File
@@ -4042,6 +4042,14 @@ dependencies = [
"uuid", "uuid",
] ]
[[package]]
name = "codex-realtime-webrtc"
version = "0.0.0"
dependencies = [
"serde",
"serde_json",
]
[[package]] [[package]]
name = "codex-response-debug-context" name = "codex-response-debug-context"
version = "0.0.0" version = "0.0.0"
@@ -4800,6 +4808,18 @@ dependencies = [
"v8", "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]] [[package]]
name = "codex-web-search-extension" name = "codex-web-search-extension"
version = "0.0.0" version = "0.0.0"
+3
View File
@@ -85,6 +85,8 @@ members = [
"network-proxy", "network-proxy",
"ollama", "ollama",
"process-hardening", "process-hardening",
"realtime-webrtc",
"voice-host",
"protocol", "protocol",
"prompts", "prompts",
"rollout", "rollout",
@@ -152,6 +154,7 @@ license = "Apache-2.0"
[workspace.dependencies] [workspace.dependencies]
# Internal # Internal
codex-realtime-webrtc = { path = "realtime-webrtc" }
app_test_support = { path = "app-server/tests/common" } app_test_support = { path = "app-server/tests/common" }
codex-analytics = { path = "analytics" } codex-analytics = { path = "analytics" }
codex-agent-graph-store = { path = "agent-graph-store" } codex-agent-graph-store = { path = "agent-graph-store" }
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "realtime-webrtc",
crate_name = "codex_realtime_webrtc",
)
+16
View File
@@ -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 }
+7
View File
@@ -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;
+63
View File
@@ -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<Vec<u8>> {
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<Option<Message>> {
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<Option<Message>> {
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)
}
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "voice-host",
crate_name = "codex_voice_host",
)
+22
View File
@@ -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"] }
+17
View File
@@ -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.
+75
View File
@@ -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")),
}
}
+123
View File
@@ -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<Child> {
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<String> {
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(())
}