From fa5ddce8d380fba6b9da049cf1f5ba2a965b7933 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Wed, 3 Jun 2026 16:55:13 -0700 Subject: [PATCH] feat: add Noise rendezvous provider API Co-authored-by: Codex noreply@openai.com --- codex-rs/core-api/src/lib.rs | 7 ++ codex-rs/exec-server/src/client.rs | 23 +++++++ codex-rs/exec-server/src/client_api.rs | 52 +++++++++++++- codex-rs/exec-server/src/client_transport.rs | 11 +++ codex-rs/exec-server/src/environment.rs | 33 +++++++++ codex-rs/exec-server/src/environment_toml.rs | 71 ++++++++++++-------- codex-rs/exec-server/src/lib.rs | 2 + 7 files changed, 171 insertions(+), 28 deletions(-) diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index 47452eb032..8d242e1b0d 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -47,7 +47,14 @@ pub use codex_core::resolve_installation_id; pub use codex_core::skills::SkillsManager; pub use codex_core::thread_store_from_config; pub use codex_exec_server::EnvironmentManager; +pub use codex_exec_server::ExecServerError; pub use codex_exec_server::ExecServerRuntimePaths; +pub use codex_exec_server::NoiseChannelIdentity; +pub use codex_exec_server::NoiseChannelPublicKey; +pub use codex_exec_server::NoiseRendezvousConnectArgs; +pub use codex_exec_server::NoiseRendezvousConnectBundle; +pub use codex_exec_server::NoiseRendezvousConnectProvider; +pub use codex_exec_server::SharedNoiseRendezvousConnectProvider; pub use codex_extension_api::empty_extension_registry; pub use codex_features::Feature; pub use codex_features::Features; diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 26b3e46e88..a7a2c7c2bb 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -23,6 +23,8 @@ use crate::ProcessId; use crate::client_api::ExecServerClientConnectOptions; use crate::client_api::ExecServerTransportParams; use crate::client_api::HttpClient; +use crate::client_api::NoiseRendezvousConnectArgs; +use crate::client_api::NoiseRendezvousConnectBundle; use crate::client_api::RemoteExecServerConnectArgs; use crate::client_api::StdioExecServerConnectArgs; use crate::connection::JsonRpcConnection; @@ -143,6 +145,26 @@ impl RemoteExecServerConnectArgs { } } +impl NoiseRendezvousConnectArgs { + /// Builds one secure connection attempt with the standard client timeouts. + /// + /// `bundle` must be freshly issued for this physical connection attempt. + pub fn new( + bundle: NoiseRendezvousConnectBundle, + harness_identity: crate::NoiseChannelIdentity, + client_name: String, + ) -> Self { + Self { + bundle, + harness_identity, + client_name, + connect_timeout: CONNECT_TIMEOUT, + initialize_timeout: INITIALIZE_TIMEOUT, + resume_session_id: None, + } + } +} + pub(crate) struct SessionState { wake_tx: watch::Sender, events: ExecProcessEventLog, @@ -237,6 +259,7 @@ impl LazyRemoteExecServerClient { if matches!( &self.transport_params, ExecServerTransportParams::WebSocketUrl { .. } + | ExecServerTransportParams::NoiseRendezvous { .. } ) => { ExecServerClient::connect_for_transport(self.transport_params.clone()).await? diff --git a/codex-rs/exec-server/src/client_api.rs b/codex-rs/exec-server/src/client_api.rs index 096a5bfc8d..776c848b77 100644 --- a/codex-rs/exec-server/src/client_api.rs +++ b/codex-rs/exec-server/src/client_api.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::path::PathBuf; +use std::sync::Arc; use std::time::Duration; use futures::future::BoxFuture; @@ -88,6 +89,23 @@ impl std::fmt::Debug for NoiseRendezvousConnectArgs { } } +/// Supplies fresh registry-authorized material for Noise rendezvous connections. +/// +/// Implementations preserve one endpoint-local harness identity while fetching +/// a fresh atomic connect bundle for every physical connection attempt. A +/// failed secure connection must remain a failure; providers must not fall back +/// to an unauthenticated transport. +pub trait NoiseRendezvousConnectProvider: Send + Sync { + /// Environment ID this provider is authorized to connect to. + fn environment_id(&self) -> &str; + + /// Returns fresh arguments for one physical connection attempt. + fn connect_args(&self) -> BoxFuture<'_, Result>; +} + +/// Shared provider used by reconnect-capable remote environments. +pub type SharedNoiseRendezvousConnectProvider = Arc; + /// Stdio connection arguments for a command-backed exec-server. #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) struct StdioExecServerConnectArgs { @@ -107,13 +125,16 @@ pub(crate) struct StdioExecServerCommand { } /// Parameters used to connect to a remote exec-server environment. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Clone)] pub(crate) enum ExecServerTransportParams { WebSocketUrl { websocket_url: String, connect_timeout: Duration, initialize_timeout: Duration, }, + NoiseRendezvous { + provider: SharedNoiseRendezvousConnectProvider, + }, #[allow(dead_code)] StdioCommand { command: StdioExecServerCommand, @@ -121,6 +142,35 @@ pub(crate) enum ExecServerTransportParams { }, } +impl std::fmt::Debug for ExecServerTransportParams { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::WebSocketUrl { + websocket_url, + connect_timeout, + initialize_timeout, + } => f + .debug_struct("WebSocketUrl") + .field("websocket_url", websocket_url) + .field("connect_timeout", connect_timeout) + .field("initialize_timeout", initialize_timeout) + .finish(), + Self::NoiseRendezvous { provider } => f + .debug_struct("NoiseRendezvous") + .field("environment_id", &provider.environment_id()) + .finish(), + Self::StdioCommand { + command, + initialize_timeout, + } => f + .debug_struct("StdioCommand") + .field("command", command) + .field("initialize_timeout", initialize_timeout) + .finish(), + } + } +} + impl ExecServerTransportParams { pub(crate) fn websocket_url(websocket_url: String) -> Self { Self::WebSocketUrl { diff --git a/codex-rs/exec-server/src/client_transport.rs b/codex-rs/exec-server/src/client_transport.rs index 65da037f27..d4298f744c 100644 --- a/codex-rs/exec-server/src/client_transport.rs +++ b/codex-rs/exec-server/src/client_transport.rs @@ -44,6 +44,17 @@ impl ExecServerClient { }) .await } + crate::client_api::ExecServerTransportParams::NoiseRendezvous { provider } => { + let args = provider.connect_args().await?; + // Keep the configured environment and the freshly authorized + // bundle bound together before opening the websocket. + if args.bundle.environment_id != provider.environment_id() { + return Err(ExecServerError::Protocol( + "Noise rendezvous provider returned a different environment id".to_string(), + )); + } + Self::connect_noise_rendezvous(args).await + } crate::client_api::ExecServerTransportParams::StdioCommand { command, initialize_timeout, diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 1dfd49d3fd..ee6aabc44b 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -9,6 +9,7 @@ use crate::ExecServerError; use crate::ExecServerRuntimePaths; use crate::ExecutorFileSystem; use crate::HttpClient; +use crate::SharedNoiseRendezvousConnectProvider; use crate::client::LazyRemoteExecServerClient; use crate::client::http_client::ReqwestHttpClient; use crate::client_api::ExecServerTransportParams; @@ -282,6 +283,37 @@ impl EnvironmentManager { .insert(environment_id, Arc::new(environment)); Ok(()) } + + /// Adds or replaces a named remote environment that connects through an + /// authenticated, end-to-end encrypted rendezvous stream. + /// + /// The provider is retained so every reconnect obtains fresh authorization. + /// This transport never falls back to the URL-only remote environment path. + pub fn upsert_noise_environment( + &self, + environment_id: String, + provider: SharedNoiseRendezvousConnectProvider, + ) -> Result<(), ExecServerError> { + if environment_id.is_empty() { + return Err(ExecServerError::Protocol( + "environment id cannot be empty".to_string(), + )); + } + if environment_id != provider.environment_id() { + return Err(ExecServerError::Protocol( + "Noise environment id does not match connection provider".to_string(), + )); + } + let environment = Environment::remote_with_transport( + ExecServerTransportParams::NoiseRendezvous { provider }, + self.local_runtime_paths.clone(), + ); + self.environments + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(environment_id, Arc::new(environment)); + Ok(()) + } } /// Concrete execution/filesystem environment selected for a session. @@ -420,6 +452,7 @@ impl Environment { websocket_url: exec_server_url, .. } => Some(exec_server_url.clone()), + ExecServerTransportParams::NoiseRendezvous { .. } => None, ExecServerTransportParams::StdioCommand { .. } => None, }; let client = LazyRemoteExecServerClient::new(remote_transport.clone()); diff --git a/codex-rs/exec-server/src/environment_toml.rs b/codex-rs/exec-server/src/environment_toml.rs index 26c178b5b8..daa024b4f5 100644 --- a/codex-rs/exec-server/src/environment_toml.rs +++ b/codex-rs/exec-server/src/environment_toml.rs @@ -48,7 +48,7 @@ struct EnvironmentToml { initialize_timeout_sec: Option, } -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Clone, Debug)] struct TomlEnvironmentProvider { default: EnvironmentDefault, include_local: bool, @@ -577,18 +577,26 @@ mod tests { ) .expect("provider"); + let ExecServerTransportParams::StdioCommand { + command, + initialize_timeout, + } = &provider.environments[0].1 + else { + panic!("expected stdio transport"); + }; assert_eq!( - provider.environments[0].1, - ExecServerTransportParams::StdioCommand { - command: StdioExecServerCommand { - program: "ssh".to_string(), - args: Vec::new(), - env: HashMap::new(), - cwd: Some(config_dir.path().join("workspace")), - }, - initialize_timeout: DEFAULT_REMOTE_EXEC_SERVER_INITIALIZE_TIMEOUT, + command, + &StdioExecServerCommand { + program: "ssh".to_string(), + args: Vec::new(), + env: HashMap::new(), + cwd: Some(config_dir.path().join("workspace")), } ); + assert_eq!( + *initialize_timeout, + DEFAULT_REMOTE_EXEC_SERVER_INITIALIZE_TIMEOUT + ); } #[test] @@ -614,26 +622,35 @@ mod tests { }) .expect("provider"); + let ExecServerTransportParams::WebSocketUrl { + websocket_url, + connect_timeout, + initialize_timeout, + } = &provider.environments[0].1 + else { + panic!("expected websocket transport"); + }; + assert_eq!(websocket_url, "ws://127.0.0.1:8765"); + assert_eq!(*connect_timeout, Duration::from_secs(12)); + assert_eq!(*initialize_timeout, Duration::from_secs(34)); + + let ExecServerTransportParams::StdioCommand { + command, + initialize_timeout, + } = &provider.environments[1].1 + else { + panic!("expected stdio transport"); + }; assert_eq!( - provider.environments[0].1, - ExecServerTransportParams::WebSocketUrl { - websocket_url: "ws://127.0.0.1:8765".to_string(), - connect_timeout: Duration::from_secs(12), - initialize_timeout: Duration::from_secs(34), - } - ); - assert_eq!( - provider.environments[1].1, - ExecServerTransportParams::StdioCommand { - command: StdioExecServerCommand { - program: "ssh".to_string(), - args: Vec::new(), - env: HashMap::new(), - cwd: None, - }, - initialize_timeout: Duration::from_secs(56), + command, + &StdioExecServerCommand { + program: "ssh".to_string(), + args: Vec::new(), + env: HashMap::new(), + cwd: None, } ); + assert_eq!(*initialize_timeout, Duration::from_secs(56)); } #[test] diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index 344a58e3b6..f60cac4f24 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -35,7 +35,9 @@ pub use client_api::ExecServerClientConnectOptions; pub use client_api::HttpClient; pub use client_api::NoiseRendezvousConnectArgs; pub use client_api::NoiseRendezvousConnectBundle; +pub use client_api::NoiseRendezvousConnectProvider; pub use client_api::RemoteExecServerConnectArgs; +pub use client_api::SharedNoiseRendezvousConnectProvider; pub use codex_file_system::CopyOptions; pub use codex_file_system::CreateDirectoryOptions; pub use codex_file_system::ExecutorFileSystem;