diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index f8eaa6234d..da4e57e05e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2910,6 +2910,7 @@ dependencies = [ "codex-utils-plugins", "dirs", "flate2", + "futures", "http 1.4.0", "libc", "pretty_assertions", diff --git a/codex-rs/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index c0ef692285..d968eec4c7 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -259,8 +259,7 @@ const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567" #[derive(Clone, Copy)] enum PluginMetricsRuntime { Classic, - Unified, - UnifiedBackground, + Unified { remote: bool, background: bool }, } fn write_curated_metrics_plugin(codex_home: &Path) -> Result { @@ -324,9 +323,12 @@ async fn assert_plugin_measurement_analytics(runtime: PluginMetricsRuntime) -> R skip_if_wine_exec!(Ok(()), "plugin metrics fixture is Unix-only"); let codex_home = TempDir::new()?; - let script_path = write_curated_metrics_plugin(codex_home.path())?; - let background = matches!(runtime, PluginMetricsRuntime::UnifiedBackground); - let unified_exec = !matches!(runtime, PluginMetricsRuntime::Classic); + let script_path = write_curated_metrics_plugin(codex_home.path())?.canonicalize()?; + let (remote, background) = match runtime { + PluginMetricsRuntime::Classic => (false, false), + PluginMetricsRuntime::Unified { remote, background } => (remote, background), + }; + let unified_exec = matches!(runtime, PluginMetricsRuntime::Unified { .. }); let mut command = vec![ "/bin/sh".to_string(), script_path.to_string_lossy().into_owned(), @@ -339,7 +341,7 @@ async fn assert_plugin_measurement_analytics(runtime: PluginMetricsRuntime) -> R PluginMetricsRuntime::Classic => { create_shell_command_sse_response(command, /*workdir*/ None, Some(5_000), call_id)? } - PluginMetricsRuntime::Unified | PluginMetricsRuntime::UnifiedBackground => { + PluginMetricsRuntime::Unified { .. } => { let arguments = serde_json::to_string(&json!({ "cmd": shlex::try_join(command.iter().map(String::as_str))?, "yield_time_ms": if background { 10 } else { 1_000 }, @@ -381,11 +383,19 @@ enabled = true )?; mount_analytics_capture(&analytics_server, codex_home.path()).await?; - let mut mcp = TestAppServer::builder() + let mut builder = TestAppServer::builder() .with_codex_home(codex_home.path()) - .without_managed_config() - .build() - .await?; + .without_managed_config(); + if remote { + builder = builder.with_exec_server_delay(Duration::ZERO); + } + let mut mcp = builder.build().await?; + if remote { + assert_eq!( + mcp.auto_env_params()?.environment_id, + codex_exec_server::REMOTE_ENVIRONMENT_ID + ); + } timeout(Duration::from_secs(10), mcp.initialize()).await??; let thread_request = mcp .send_thread_start_request_with_auto_env(ThreadStartParams { @@ -576,11 +586,40 @@ async fn classic_plugin_script_emits_measurement_analytics() -> Result<()> { #[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn unified_plugin_script_emits_measurement_analytics() -> Result<()> { - assert_plugin_measurement_analytics(PluginMetricsRuntime::Unified).await + assert_plugin_measurement_analytics(PluginMetricsRuntime::Unified { + remote: false, + background: false, + }) + .await +} + +#[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn remote_unified_plugin_script_emits_measurement_analytics() -> Result<()> { + assert_plugin_measurement_analytics(PluginMetricsRuntime::Unified { + remote: true, + background: false, + }) + .await } #[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn unified_background_plugin_script_emits_measurements_after_turn_completion() -> Result<()> { - assert_plugin_measurement_analytics(PluginMetricsRuntime::UnifiedBackground).await + assert_plugin_measurement_analytics(PluginMetricsRuntime::Unified { + remote: false, + background: true, + }) + .await +} + +#[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn remote_unified_background_plugin_script_emits_measurements_after_turn_completion() +-> Result<()> { + assert_plugin_measurement_analytics(PluginMetricsRuntime::Unified { + remote: true, + background: true, + }) + .await } diff --git a/codex-rs/core-plugins/Cargo.toml b/codex-rs/core-plugins/Cargo.toml index 931497086e..0899163acc 100644 --- a/codex-rs/core-plugins/Cargo.toml +++ b/codex-rs/core-plugins/Cargo.toml @@ -38,6 +38,7 @@ codex-utils-plugins = { workspace = true } chrono = { workspace = true } dirs = { workspace = true } flate2 = { workspace = true } +futures = { workspace = true } http = { workspace = true } regex = { workspace = true } semver = { workspace = true } diff --git a/codex-rs/core-plugins/src/plugin_metrics_sidecar.rs b/codex-rs/core-plugins/src/plugin_metrics_sidecar.rs index 3b0b4d43a1..aefab13ecf 100644 --- a/codex-rs/core-plugins/src/plugin_metrics_sidecar.rs +++ b/codex-rs/core-plugins/src/plugin_metrics_sidecar.rs @@ -1,8 +1,15 @@ use crate::ResolvedPluginMetricsOperation; use codex_analytics::PluginMeasurementRow; +use codex_exec_server::Environment; +use codex_exec_server::ExecutorFileSystem; +use codex_exec_server::FileSystemReadStream; +use codex_exec_server::RemoveOptions; use codex_protocol::models::AdditionalPermissionProfile; use codex_protocol::models::FileSystemPermissions; use codex_utils_absolute_path::AbsolutePathBuf; +use codex_utils_path_uri::PathConvention; +use codex_utils_path_uri::PathUri; +use futures::StreamExt; use serde::Deserialize; use std::collections::BTreeMap; use std::collections::BTreeSet; @@ -10,6 +17,7 @@ use std::collections::HashMap; use std::io::Read; use std::io::Seek; use std::io::SeekFrom; +use std::sync::Arc; use tempfile::NamedTempFile; use uuid::Uuid; @@ -26,14 +34,51 @@ pub struct PluginMeasurementBatch { } pub struct PluginMetricsSidecar { - output_file: NamedTempFile, - _output_dir: tempfile::TempDir, + output: PluginMetricsOutput, absolute_output_dir: AbsolutePathBuf, output_env_value: String, resolved: ResolvedPluginMetricsOperation, execution_id: String, } +enum PluginMetricsOutput { + Local { + file: NamedTempFile, + _directory: tempfile::TempDir, + }, + Remote { + file_stream: tokio::sync::Mutex, + _directory: RemotePluginMetricsDirectory, + }, +} + +struct RemotePluginMetricsDirectory { + filesystem: Arc, + path: PathUri, +} + +impl Drop for RemotePluginMetricsDirectory { + fn drop(&mut self) { + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + return; + }; + let filesystem = Arc::clone(&self.filesystem); + let path = self.path.clone(); + runtime.spawn(async move { + let _ = filesystem + .remove( + &path, + RemoveOptions { + recursive: true, + force: true, + }, + /*sandbox*/ None, + ) + .await; + }); + } +} + #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct OutputEnvelope { @@ -65,8 +110,10 @@ impl PluginMetricsSidecar { let absolute_output_path = AbsolutePathBuf::from_absolute_path(output_file.path()).ok()?; let output_env_value = absolute_output_path.as_path().to_str()?.to_string(); Some(Self { - output_file, - _output_dir: sidecar_dir, + output: PluginMetricsOutput::Local { + file: output_file, + _directory: sidecar_dir, + }, absolute_output_dir, output_env_value, resolved, @@ -74,6 +121,56 @@ impl PluginMetricsSidecar { }) } + pub async fn create_remote( + environment: &Environment, + resolved: ResolvedPluginMetricsOperation, + ) -> Option { + let temp_dir = environment.info().await.ok()?.temp_dir?; + // Permission overlays still use host-native AbsolutePathBuf roots, so a + // foreign executor path cannot be granted its exact sidecar directory. + if !cfg!(unix) || temp_dir.infer_path_convention() != Some(PathConvention::Posix) { + tracing::debug!( + executor_temp_dir = %temp_dir, + "plugin metrics require POSIX executor paths on a POSIX frontend" + ); + return None; + } + let execution_id = Uuid::new_v4().to_string(); + let directory_path = temp_dir + .join(&format!("codex-plugin-metrics-{execution_id}")) + .ok()?; + let absolute_output_dir = directory_path.to_abs_path().ok()?; + environment + .create_private_directory(&directory_path) + .await + .ok()?; + let directory = RemotePluginMetricsDirectory { + filesystem: environment.get_filesystem(), + path: directory_path, + }; + let output_path = directory.path.join("measurements.json").ok()?; + directory + .filesystem + .write_file(&output_path, Vec::new(), /*sandbox*/ None) + .await + .ok()?; + let file_stream = directory + .filesystem + .read_file_stream(&output_path, /*sandbox*/ None) + .await + .ok()?; + Some(Self { + output: PluginMetricsOutput::Remote { + file_stream: tokio::sync::Mutex::new(file_stream), + _directory: directory, + }, + absolute_output_dir, + output_env_value: output_path.inferred_native_path_string(), + resolved, + execution_id, + }) + } + pub fn install_output_env(&self, env: &mut HashMap) { env.insert( PLUGIN_METRICS_OUTPUT_ENV_VAR.to_string(), @@ -83,7 +180,7 @@ impl PluginMetricsSidecar { #[cfg(test)] fn absolute_output_path(&self) -> AbsolutePathBuf { - AbsolutePathBuf::from_absolute_path(self.output_file.path()).expect("absolute output path") + AbsolutePathBuf::from_absolute_path(&self.output_env_value).expect("absolute output path") } pub fn additional_permissions(&self) -> AdditionalPermissionProfile { @@ -96,11 +193,32 @@ impl PluginMetricsSidecar { } } - pub fn finish(mut self, exit_code: i32) -> Option { + pub async fn finish(mut self, exit_code: i32) -> Option { if exit_code != 0 { return None; } - let rows = parse_output(self.output_file.as_file_mut(), &self.resolved)?; + let mut contents = Vec::new(); + match &mut self.output { + PluginMetricsOutput::Local { file, .. } => { + let output_file = file.as_file_mut(); + output_file.seek(SeekFrom::Start(0)).ok()?; + output_file + .take(MAX_OUTPUT_BYTES + 1) + .read_to_end(&mut contents) + .ok()?; + } + PluginMetricsOutput::Remote { file_stream, .. } => { + let file_stream = file_stream.get_mut(); + while let Some(chunk) = file_stream.next().await { + let chunk = chunk.ok()?; + if contents.len().saturating_add(chunk.len()) > MAX_OUTPUT_BYTES as usize { + return None; + } + contents.extend_from_slice(&chunk); + } + } + } + let rows = parse_output(&contents, &self.resolved)?; (!rows.is_empty()).then(|| PluginMeasurementBatch { plugin_id: self.resolved.plugin_id.as_key(), execution_id: self.execution_id, @@ -119,19 +237,13 @@ pub fn strip_output_env(env: &mut HashMap) { } fn parse_output( - output_file: &mut std::fs::File, + contents: &[u8], resolved: &ResolvedPluginMetricsOperation, ) -> Option> { - let mut contents = Vec::new(); - output_file.seek(SeekFrom::Start(0)).ok()?; - output_file - .take(MAX_OUTPUT_BYTES + 1) - .read_to_end(&mut contents) - .ok()?; if contents.len() as u64 > MAX_OUTPUT_BYTES { return None; } - let output: OutputEnvelope = serde_json::from_slice(&contents).ok()?; + let output: OutputEnvelope = serde_json::from_slice(contents).ok()?; if output.version != 1 || output.measurements.len() > MAX_OUTPUT_ROWS { return None; } diff --git a/codex-rs/core-plugins/src/plugin_metrics_sidecar_tests.rs b/codex-rs/core-plugins/src/plugin_metrics_sidecar_tests.rs index 69b8199aa7..4be61b64f2 100644 --- a/codex-rs/core-plugins/src/plugin_metrics_sidecar_tests.rs +++ b/codex-rs/core-plugins/src/plugin_metrics_sidecar_tests.rs @@ -63,8 +63,8 @@ fn resolved_operation() -> ResolvedPluginMetricsOperation { } } -#[test] -fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() { +#[tokio::test] +async fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() { let sidecar = create_sidecar(); let path = sidecar.absolute_output_path(); std::fs::write( @@ -87,7 +87,10 @@ fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() { ) .expect("write output"); - let batch = sidecar.finish(/*exit_code*/ 0).expect("valid measurements"); + let batch = sidecar + .finish(/*exit_code*/ 0) + .await + .expect("valid measurements"); let execution_id = batch.execution_id.clone(); assert_eq!( batch, @@ -118,8 +121,8 @@ fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() { assert!(!path.exists()); } -#[test] -fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() { +#[tokio::test] +async fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() { for output in [ r#"{"version":2,"measurements":[]}"#.as_bytes().to_vec(), r#"{"version":1,"measurements":[],"unknown":true}"#.as_bytes().to_vec(), @@ -134,7 +137,7 @@ fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() { let sidecar = create_sidecar(); let path = sidecar.absolute_output_path(); std::fs::write(path.as_path(), output).expect("write output"); - assert_eq!(sidecar.finish(/*exit_code*/ 0), None); + assert_eq!(sidecar.finish(/*exit_code*/ 0).await, None); assert!(!path.exists()); } @@ -145,7 +148,7 @@ fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() { r#"{"version":1,"measurements":[{"name":"files_scanned","value":1}]}"#, ) .expect("write output"); - assert_eq!(sidecar.finish(/*exit_code*/ 1), None); + assert_eq!(sidecar.finish(/*exit_code*/ 1).await, None); assert!(!path.exists()); } @@ -176,8 +179,8 @@ fn reserved_output_env_is_absent_without_sidecar_and_cannot_be_overridden() { } #[cfg(unix)] -#[test] -fn sidecar_reads_the_original_file_after_path_replacement() { +#[tokio::test] +async fn sidecar_reads_the_original_file_after_path_replacement() { let sidecar = create_sidecar(); let path = sidecar.absolute_output_path(); std::fs::remove_file(path.as_path()).expect("remove original output path"); @@ -187,6 +190,6 @@ fn sidecar_reads_the_original_file_after_path_replacement() { ) .expect("write replacement output"); - assert_eq!(sidecar.finish(/*exit_code*/ 0), None); + assert_eq!(sidecar.finish(/*exit_code*/ 0).await, None); assert!(!path.exists()); } diff --git a/codex-rs/core-plugins/src/script_attribution.rs b/codex-rs/core-plugins/src/script_attribution.rs index 509af95db1..1b61d3c138 100644 --- a/codex-rs/core-plugins/src/script_attribution.rs +++ b/codex-rs/core-plugins/src/script_attribution.rs @@ -208,6 +208,13 @@ impl TrustedPluginRoots { cwd: &AbsolutePathBuf, ) -> Option { let attribution = self.resolve_attribution(command, cwd)?; + self.metrics_operation_for_attribution(attribution) + } + + fn metrics_operation_for_attribution( + &self, + attribution: PluginCommandAttribution, + ) -> Option { let mut matches = self.roots.iter().filter_map(|root| { (root.plugin_id == attribution.plugin_id) .then(|| { @@ -262,6 +269,19 @@ impl TrustedPluginRoots { (contents == candidate.contents).then_some(candidate.attribution) } + /// Resolves one trusted executor script to one manifest-declared operation. + pub async fn resolve_metrics_operation_in_filesystem( + &self, + command: &[String], + cwd: &PathUri, + file_system: &dyn ExecutorFileSystem, + ) -> Option { + let attribution = self + .resolve_executor_attribution(command, cwd, file_system) + .await?; + self.metrics_operation_for_attribution(attribution) + } + fn local_candidate_for_executor_script( &self, script: &PathUri, diff --git a/codex-rs/core/src/plugins/metrics.rs b/codex-rs/core/src/plugins/metrics.rs index 04d8432179..5e81763249 100644 --- a/codex-rs/core/src/plugins/metrics.rs +++ b/codex-rs/core/src/plugins/metrics.rs @@ -1,10 +1,35 @@ use crate::session::session::Session; use crate::session::turn_context::TurnContext; +use crate::tools::sandboxing::ToolCtx; use codex_analytics::PluginMeasurementsInput; use codex_core_plugins::PluginMetricsSidecar; +use codex_exec_server::Environment; +use codex_utils_path_uri::PathUri; + +/// Creates a metrics sidecar for one plugin command. +pub(crate) async fn sidecar_for_command( + ctx: &ToolCtx, + command: &[String], + cwd: &PathUri, + environment: &Environment, +) -> Option { + if !ctx.session.services.analytics_events_client.is_enabled() { + return None; + } + let resolved = ctx + .step_context + .turn + .plugin_metrics_operation_for_command(command, cwd, environment) + .await?; + if environment.is_remote() { + PluginMetricsSidecar::create_remote(environment, resolved).await + } else { + PluginMetricsSidecar::create(resolved) + } +} /// Finishes a metrics sidecar and publishes any valid rows. -pub(crate) fn finish_and_track_measurements( +pub(crate) async fn finish_and_track_measurements( metrics_sidecar: Option, exit_code: i32, session: &Session, @@ -14,7 +39,7 @@ pub(crate) fn finish_and_track_measurements( let Some(metrics_sidecar) = metrics_sidecar else { return; }; - let Some(batch) = metrics_sidecar.finish(exit_code) else { + let Some(batch) = metrics_sidecar.finish(exit_code).await else { return; }; session diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index e1d282c28d..eb8cf98608 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -243,14 +243,24 @@ impl TurnContext { } } - pub(crate) fn plugin_metrics_operation_for_command( + pub(crate) async fn plugin_metrics_operation_for_command( &self, command: &[String], - cwd: &AbsolutePathBuf, + cwd: &PathUri, + environment: &Environment, ) -> Option { - self.extension_data - .get::()? - .resolve_metrics_operation(command, cwd) + let trusted_roots = self.extension_data.get::()?; + if environment.is_remote() { + trusted_roots + .resolve_metrics_operation_in_filesystem( + command, + cwd, + environment.get_filesystem().as_ref(), + ) + .await + } else { + trusted_roots.resolve_metrics_operation(command, &cwd.to_abs_path().ok()?) + } } pub(crate) fn permission_profile(&self) -> PermissionProfile { diff --git a/codex-rs/core/src/tools/runtimes/shell.rs b/codex-rs/core/src/tools/runtimes/shell.rs index 00e0c3e503..623bafb536 100644 --- a/codex-rs/core/src/tools/runtimes/shell.rs +++ b/codex-rs/core/src/tools/runtimes/shell.rs @@ -11,6 +11,7 @@ pub(crate) mod zsh_fork_backend; use crate::exec::ExecCapturePolicy; use crate::guardian::GuardianNetworkAccessTrigger; use crate::plugins::metrics::finish_and_track_measurements; +use crate::plugins::metrics::sidecar_for_command; use crate::sandboxing::ExecOptions; use crate::sandboxing::SandboxPermissions; use crate::sandboxing::execute_env; @@ -212,15 +213,14 @@ impl ToolRuntime for ShellRuntime { managed_network_for_sandbox_permissions(req.network.as_ref(), sandbox_permissions); let mut env = exec_env_for_sandbox_permissions(&req.env, sandbox_permissions); let explicit_env_overrides = req.explicit_env_overrides.clone(); - let metrics_sidecar = (!req.turn_environment.environment.is_remote() - && ctx.session.services.analytics_events_client.is_enabled()) - .then(|| { - ctx.step_context - .turn - .plugin_metrics_operation_for_command(&req.command, &req.cwd) - }) - .flatten() - .and_then(PluginMetricsSidecar::create); + let cwd = PathUri::from_abs_path(&req.cwd); + let metrics_sidecar = sidecar_for_command( + ctx, + &req.command, + &cwd, + req.turn_environment.environment.as_ref(), + ) + .await; if let Some(sidecar) = metrics_sidecar.as_ref() { sidecar.install_output_env(&mut env); } @@ -320,7 +320,8 @@ impl ToolRuntime for ShellRuntime { &ctx.session, &ctx.step_context.turn, &ctx.call_id, - ); + ) + .await; Ok(out) } } diff --git a/codex-rs/core/src/tools/runtimes/unified_exec.rs b/codex-rs/core/src/tools/runtimes/unified_exec.rs index ef5190a76c..94829bd819 100644 --- a/codex-rs/core/src/tools/runtimes/unified_exec.rs +++ b/codex-rs/core/src/tools/runtimes/unified_exec.rs @@ -9,6 +9,7 @@ use crate::exec::ExecExpiration; use crate::guardian::GUARDIAN_REVIEW_TIMEOUT; use crate::guardian::GuardianNetworkAccessTrigger; use crate::guardian::routes_approval_to_guardian; +use crate::plugins::metrics::sidecar_for_command; use crate::sandboxing::ExecOptions; use crate::sandboxing::ExecServerEnvConfig; use crate::sandboxing::SandboxPermissions; @@ -337,16 +338,13 @@ impl<'a> ToolRuntime for UnifiedExecRunt None => (env, None, None), }; let explicit_env_overrides = req.explicit_env_overrides.clone(); - let metrics_sidecar = (!environment_is_remote - && ctx.session.services.analytics_events_client.is_enabled()) - .then(|| { - let cwd = req.cwd.to_abs_path().ok()?; - ctx.step_context - .turn - .plugin_metrics_operation_for_command(&req.command, &cwd) - }) - .flatten() - .and_then(PluginMetricsSidecar::create); + let metrics_sidecar = sidecar_for_command( + ctx, + &req.command, + &req.cwd, + req.turn_environment.environment.as_ref(), + ) + .await; if let Some(sidecar) = metrics_sidecar.as_ref() { sidecar.install_output_env(&mut env); } diff --git a/codex-rs/core/src/unified_exec/async_watcher.rs b/codex-rs/core/src/unified_exec/async_watcher.rs index 650d5dfd75..f846d15737 100644 --- a/codex-rs/core/src/unified_exec/async_watcher.rs +++ b/codex-rs/core/src/unified_exec/async_watcher.rs @@ -216,7 +216,8 @@ pub(crate) fn spawn_exit_watcher( &session_ref, &turn_ref, &call_id, - ); + ) + .await; emit_exec_end_for_unified_exec( session_ref, turn_ref, diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 13292c649c..6b12ea2c52 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -251,14 +251,15 @@ struct InitialExecCommandGuard { } impl InitialExecCommandGuard { - fn finish_plugin_metrics(&mut self, context: &UnifiedExecContext, exit_code: i32) { + async fn finish_plugin_metrics(&mut self, context: &UnifiedExecContext, exit_code: i32) { finish_and_track_measurements( self.metrics_sidecar.take(), exit_code, &context.session, &context.step_context.turn, &context.call_id, - ); + ) + .await; } } @@ -661,7 +662,8 @@ impl UnifiedExecProcessManager { &context.session, &context.step_context.turn, &context.call_id, - ); + ) + .await; (None, exit_code) } ProcessStatus::Unknown => { @@ -694,7 +696,9 @@ impl UnifiedExecProcessManager { } let exit_code = process.exit_code(); let exit = exit_code.unwrap_or(-1); - initial_exec_command_guard.finish_plugin_metrics(context, exit); + initial_exec_command_guard + .finish_plugin_metrics(context, exit) + .await; emit_exec_end_for_unified_exec( Arc::clone(&context.session), Arc::clone(&context.step_context.turn), diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index 5a342c5152..abbdf16305 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -97,6 +97,9 @@ pub struct EnvironmentInfo { /// On Windows, a command's `TEMP` or `TMP` overrides take precedence. #[serde(default, skip_serializing_if = "Option::is_none")] pub temporary_directories: Option>, + /// Executor-native temporary directory for private, child-visible sidecars. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub temp_dir: Option, /// Optional executor features that clients must gate before sending newer request fields. #[serde(default)] pub capabilities: EnvironmentCapabilities, @@ -145,30 +148,33 @@ impl EnvironmentInfo { } else { &["TMPDIR"] }; + let normalize_temp_path = |path: std::ffi::OsString| { + PathUri::from_host_native_path(&path).ok().or_else(|| { + if cfg!(unix) { + PathUri::from_host_native_path(cwd.as_ref()?.join(path)).ok() + } else { + None + } + }) + }; let mut temporary_directories = Vec::new(); for name in temporary_directory_env_vars { if let Some(path) = std::env::var_os(name) .filter(|path| !path.is_empty()) .filter(|path| cfg!(unix) || std::path::Path::new(path).is_absolute()) - .and_then(|path| { - PathUri::from_host_native_path(&path).ok().or_else(|| { - if cfg!(unix) { - PathUri::from_host_native_path(cwd.as_ref()?.join(path)).ok() - } else { - None - } - }) - }) + .and_then(&normalize_temp_path) && !temporary_directories.contains(&path) { temporary_directories.push(path); } } + let temp_dir = normalize_temp_path(std::env::temp_dir().into_os_string()); Self { shell: codex_shell_command::shell_detect::default_user_shell().into(), cwd: cwd.and_then(|cwd| PathUri::from_host_native_path(cwd).ok()), temporary_directories: Some(temporary_directories), + temp_dir, capabilities: EnvironmentCapabilities { network_proxy_launch: true, capability_discovery_sandbox: true, @@ -417,6 +423,9 @@ pub struct FsCreateDirectoryParams { pub path: PathUri, pub recursive: Option, pub sandbox: Option, + /// Atomically restrict a newly created, non-recursive directory to its owner. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub private: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -902,6 +911,7 @@ mod tests { }, cwd: None, temporary_directories: None, + temp_dir: None, capabilities: EnvironmentCapabilities::default(), } ); @@ -1000,10 +1010,9 @@ mod tests { .join("relative-temp"), ) .expect("absolute temporary directory URI"); - assert_eq!( - EnvironmentInfo::local().temporary_directories, - Some(vec![expected]) - ); + let info = EnvironmentInfo::local(); + assert_eq!(info.temporary_directories, Some(vec![expected.clone()])); + assert_eq!(info.temp_dir, Some(expected)); } #[test] diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 32a669ecef..7616692486 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -37,9 +37,11 @@ use crate::local_file_system::LocalFileSystem; use crate::local_process::LocalProcess; use crate::process::ExecBackend; use crate::protocol::EnvironmentInfo; +use crate::protocol::FsCreateDirectoryParams; use crate::remote::NoiseRendezvousEnvironmentConfig; use crate::remote_file_system::RemoteFileSystem; use crate::remote_process::RemoteProcess; +use codex_utils_path_uri::PathUri; use tokio::sync::watch; use tokio_util::task::AbortOnDropHandle; @@ -897,6 +899,26 @@ impl Environment { } } + /// Atomically creates an owner-private directory on a remote executor. + pub async fn create_private_directory(&self, path: &PathUri) -> Result<(), ExecServerError> { + let Some(client) = &self.remote_client else { + return Err(ExecServerError::Protocol( + "private executor directory creation requires a remote environment".to_string(), + )); + }; + client + .get() + .await? + .fs_create_directory(FsCreateDirectoryParams { + path: path.clone(), + recursive: Some(false), + sandbox: None, + private: Some(true), + }) + .await?; + Ok(()) + } + /// Discovers plugin and skill manifests through the environment's high-level discovery API. pub async fn discover_capability_roots( &self, @@ -1078,6 +1100,10 @@ mod tests { .expect("cwd URI") ) ); + assert_eq!( + info.temp_dir, + PathUri::from_host_native_path(std::env::temp_dir()).ok() + ); } #[tokio::test] diff --git a/codex-rs/exec-server/src/remote_file_system.rs b/codex-rs/exec-server/src/remote_file_system.rs index dfde60b8c2..ed51e0bb5e 100644 --- a/codex-rs/exec-server/src/remote_file_system.rs +++ b/codex-rs/exec-server/src/remote_file_system.rs @@ -144,6 +144,7 @@ impl RemoteFileSystem { path: path.clone(), recursive: Some(options.recursive), sandbox: remote_sandbox_context(sandbox), + private: None, }) .await; self.metadata_requests.lock().await.clear(); diff --git a/codex-rs/exec-server/src/sandboxed_file_system.rs b/codex-rs/exec-server/src/sandboxed_file_system.rs index 5feefe3bad..8bfa8fea11 100644 --- a/codex-rs/exec-server/src/sandboxed_file_system.rs +++ b/codex-rs/exec-server/src/sandboxed_file_system.rs @@ -138,6 +138,7 @@ impl SandboxedFileSystem { path: path.clone(), recursive: Some(options.recursive), sandbox: None, + private: None, }), ) .await? diff --git a/codex-rs/exec-server/src/server/file_system_handler.rs b/codex-rs/exec-server/src/server/file_system_handler.rs index 7ead650165..23151c2e5f 100644 --- a/codex-rs/exec-server/src/server/file_system_handler.rs +++ b/codex-rs/exec-server/src/server/file_system_handler.rs @@ -153,6 +153,27 @@ impl FileSystemHandler { &self, params: FsCreateDirectoryParams, ) -> Result { + if params.private.unwrap_or(false) { + if params.recursive.unwrap_or(false) || params.sandbox.is_some() { + return Err(invalid_request( + "private directories must be non-recursive and unsandboxed".to_string(), + )); + } + #[cfg(unix)] + { + let path = params.path.to_abs_path().map_err(map_fs_error)?; + let mut builder = tokio::fs::DirBuilder::new(); + builder.mode(0o700); + builder.create(path.as_path()).await.map_err(map_fs_error)?; + return Ok(FsCreateDirectoryResponse {}); + } + #[cfg(not(unix))] + { + return Err(invalid_request( + "owner-private directories are unsupported on this platform".to_string(), + )); + } + } let recursive = params.recursive.unwrap_or(true); self.file_system .create_directory( @@ -289,6 +310,9 @@ fn map_fs_error(err: io::Error) -> JSONRPCErrorError { #[cfg(test)] mod tests { + #[cfg(unix)] + use std::os::unix::fs::PermissionsExt; + use codex_protocol::protocol::NetworkAccess; use codex_protocol::protocol::SandboxPolicy; use codex_utils_path_uri::PathUri; @@ -299,6 +323,62 @@ mod tests { use crate::protocol::FsReadFileParams; use crate::protocol::FsWriteFileParams; + #[cfg(unix)] + #[tokio::test] + async fn private_directories_are_created_with_owner_only_permissions() { + let temp_dir = tempfile::tempdir().expect("tempdir"); + let runtime_paths = ExecServerRuntimePaths::new( + std::env::current_exe().expect("current exe"), + /*codex_linux_sandbox_exe*/ None, + ) + .expect("runtime paths"); + let handler = FileSystemHandler::new(runtime_paths); + let directory = temp_dir.path().join("private-metrics"); + handler + .create_directory(FsCreateDirectoryParams { + path: PathUri::from_host_native_path(&directory).expect("directory URI"), + recursive: Some(false), + sandbox: None, + private: Some(true), + }) + .await + .expect("create private directory"); + + assert_eq!( + std::fs::metadata(directory) + .expect("directory metadata") + .permissions() + .mode() + & 0o777, + 0o700 + ); + } + + #[cfg(windows)] + #[tokio::test] + async fn private_directories_are_rejected_when_owner_only_permissions_are_unsupported() { + let temp_dir = tempfile::tempdir().expect("tempdir"); + let runtime_paths = ExecServerRuntimePaths::new( + std::env::current_exe().expect("current exe"), + /*codex_linux_sandbox_exe*/ None, + ) + .expect("runtime paths"); + let handler = FileSystemHandler::new(runtime_paths); + let directory = temp_dir.path().join("private-metrics"); + let error = handler + .create_directory(FsCreateDirectoryParams { + path: PathUri::from_host_native_path(&directory).expect("directory URI"), + recursive: Some(false), + sandbox: None, + private: Some(true), + }) + .await + .expect_err("private directories must fail closed"); + + assert!(error.message.contains("owner-private directories")); + assert!(!directory.exists()); + } + #[tokio::test] async fn no_platform_sandbox_policies_do_not_require_configured_sandbox_helper() { let temp_dir = tempfile::tempdir().expect("tempdir"); diff --git a/codex-rs/exec-server/tests/process.rs b/codex-rs/exec-server/tests/process.rs index 1b0289776a..34dab559ec 100644 --- a/codex-rs/exec-server/tests/process.rs +++ b/codex-rs/exec-server/tests/process.rs @@ -211,6 +211,8 @@ async fn exec_server_runs_ordinary_requests_serially_by_default() -> anyhow::Res expected_environment_info.temporary_directories = Some(vec![PathUri::from_host_native_path( temporary_directory.path(), )?]); + expected_environment_info.temp_dir = + Some(PathUri::from_host_native_path(temporary_directory.path())?); assert_eq!( serde_json::from_value::(result)?, expected_environment_info