Initialize the packaged GStreamer runtime in the voice host (#42631)

## Why

The voice helper previously supported only handshake and shutdown, without
validating that its packaged native runtime could be initialized safely.

## What changed

- Add an `initializeRuntime` protocol exchange that loads GStreamer and the
  required plugins from physical package paths without opening audio devices.
- Restrict plugin discovery, registry access, and native library search paths,
  and keep loaded libraries alive until the helper exits.
- Give initialization a dedicated deadline and terminate the owned helper when
  initialization is cancelled or fails.
- Allow binary-only Rust targets to disable the default Bazel library target.

## Testing

- Cover helper-only packages, cancellation, and environment filtering.
- Add an ignored integration test for initialization from a relocated prepared
  runtime and rejection of duplicate initialization.

GitOrigin-RevId: 8fff68fe26e52cb2e0722dc98fb5f124eddb0d4d
This commit is contained in:
Benjamin Carlsson authored and copyberry committed 2026-09-03 22:53:45 +00:00
1 parent 280ae8b9fc
commit d979df154c
13 files changed
+386 -16

No files matched your search

+28 -3
View File
@@ -17,6 +17,7 @@ use crate::encode_frame;
use crate::message_reader::MessageReader;
const DEADLINE: Duration = Duration::from_secs(/*secs*/ 5);
const RUNTIME_INITIALIZATION_DEADLINE: Duration = Duration::from_secs(/*secs*/ 30);
/// Owns one helper. Dropping it terminates the process and leaves its waiter to reap it.
/// A successful handshake establishes compatibility only, not an active audio session.
@@ -27,6 +28,17 @@ pub struct VoiceHost {
}
impl VoiceHost {
/// Initialize the packaged native runtime without opening devices or starting a session.
pub async fn initialize_runtime(mut self) -> Result<Self> {
self.exchange(
Message::InitializeRuntime {},
Message::RuntimeReady {},
RUNTIME_INITIALIZATION_DEADLINE,
)
.await?;
Ok(self)
}
pub async fn connect(package: &CodexPackageLayout, build_commit: &str) -> Result<Self> {
let root = package.package_dir.as_path().canonicalize()?;
let name = if cfg!(windows) {
@@ -66,13 +78,16 @@ impl VoiceHost {
build_commit: build_commit.to_owned(),
},
Message::Ready {},
DEADLINE,
)
.await?;
Ok(host)
}
pub async fn close(mut self) -> Result<()> {
let result = self.exchange(Message::Close {}, Message::Closed {}).await;
let result = self
.exchange(Message::Close {}, Message::Closed {}, DEADLINE)
.await;
if result.is_err() {
self.process.terminate();
}
@@ -82,8 +97,13 @@ impl VoiceHost {
Ok(())
}
async fn exchange(&mut self, request: Message, expected: Message) -> Result<()> {
timeout(DEADLINE, async {
async fn exchange(
&mut self,
request: Message,
expected: Message,
deadline: Duration,
) -> Result<()> {
timeout(deadline, async {
self.process
.writer_sender()
.send(encode_frame(&request)?)
@@ -128,6 +148,11 @@ fn child_environment(vars: impl Iterator<Item = (OsString, OsString)>) -> HashMa
.then(|| Some((key, value.into_string().ok()?)))
.flatten()
})
.chain(
crate::RUNTIME_ENVIRONMENT
.into_iter()
.map(|(key, value)| (key.to_owned(), value.to_owned())),
)
.collect()
}
@@ -14,6 +14,8 @@ fn forwards_only_explicit_device_network_and_os_inputs() {
("DYLD_INSERT_LIBRARIES", "loader"),
("PATH", "project"),
("GST_PLUGIN_PATH", "plugins"),
("GST_REGISTRY", "untrusted-registry"),
("GST_REGISTRY_FORK", "yes"),
("OPENAI_API_KEY", "secret"),
];
assert_eq!(
@@ -25,6 +27,37 @@ fn forwards_only_explicit_device_network_and_os_inputs() {
input[..3]
.iter()
.map(|(key, value)| (key.to_string(), value.to_string()))
.chain(crate::RUNTIME_ENVIRONMENT.map(|(key, value)| (key.into(), value.into())))
.collect()
);
}
#[cfg(unix)]
#[tokio::test]
async fn cancelling_initialization_terminates_the_owned_helper() -> anyhow::Result<()> {
let spawned = codex_utils_pty::spawn_pipe_process(
std::path::Path::new("/bin/sleep"),
&["30".to_owned()],
std::path::Path::new("/"),
&child_environment(std::iter::empty()),
/*arg0*/ &None,
&[],
)
.await?;
drop(spawned.stderr_rx);
let (_, unused_exit) = tokio::sync::oneshot::channel();
let host = super::VoiceHost {
process: spawned.session,
output: crate::message_reader::MessageReader::new(spawned.stdout_rx),
exit: unused_exit,
};
let mut initialization = Box::pin(host.initialize_runtime());
std::future::poll_fn(|context| {
assert!(std::future::Future::poll(initialization.as_mut(), context).is_pending());
std::task::Poll::Ready(())
})
.await;
drop(initialization);
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 1), spawned.exit_rx).await??;
Ok(())
}
+1
View File
@@ -5,6 +5,7 @@ mod protocol;
pub use client::VoiceHost;
pub use protocol::MAX_FRAME_BYTES;
pub use protocol::Message;
pub use protocol::RUNTIME_ENVIRONMENT;
pub use protocol::decode_frame;
pub use protocol::encode_frame;
pub use protocol::read_message;
+16
View File
@@ -8,6 +8,20 @@ use serde::Serialize;
pub const MAX_FRAME_BYTES: usize = 256;
/// Fixed child settings prevent native initialization from scanning system plugins or caches.
pub const RUNTIME_ENVIRONMENT: [(&str, &str); 7] = [
("GST_PLUGIN_PATH", ""),
("GST_PLUGIN_PATH_1_0", ""),
("GST_PLUGIN_SYSTEM_PATH", ""),
("GST_PLUGIN_SYSTEM_PATH_1_0", ""),
(
"GST_REGISTRY",
if cfg!(windows) { "NUL" } else { "/dev/null" },
),
("GST_REGISTRY_UPDATE", "no"),
("GST_REGISTRY_FORK", "no"),
];
#[derive(Debug, Deserialize, PartialEq, Serialize)]
#[serde(
tag = "type",
@@ -18,6 +32,8 @@ pub const MAX_FRAME_BYTES: usize = 256;
pub enum Message {
Hello { protocol: u32, build_commit: String },
Ready {},
InitializeRuntime {},
RuntimeReady {},
Close {},
Closed {},
}