Collect plugin metrics from remote executors (#38283)

## What changed

- Resolve manifest-declared metric operations against the executor filesystem for remote plugin commands.
- Create the measurement sidecar in an executor-native, owner-private temporary directory, stream its bounded output back for validation, and clean up the directory afterward.
- Extend the exec-server protocol with the executor temporary directory and atomic private-directory creation.

## Testing

- Cover remote unified-exec measurements for foreground and background commands.
- Verify private directories use owner-only permissions on Unix and fail closed on unsupported platforms.

GitOrigin-RevId: dbbd0a84717b91237fc5728e510e18994eb46dd4
This commit is contained in:
Kyle Brown authored and copyberry committed 2026-08-13 00:39:04 +00:00
1 parent 1e71e35df6
commit 9579479d28
18 files changed
+416 -82

No files matched your search

+1
View File
@@ -2910,6 +2910,7 @@ dependencies = [
"codex-utils-plugins", "codex-utils-plugins",
"dirs", "dirs",
"flate2", "flate2",
"futures",
"http 1.4.0", "http 1.4.0",
"libc", "libc",
"pretty_assertions", "pretty_assertions",
+51 -12
View File
@@ -259,8 +259,7 @@ const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567"
#[derive(Clone, Copy)] #[derive(Clone, Copy)]
enum PluginMetricsRuntime { enum PluginMetricsRuntime {
Classic, Classic,
Unified, Unified { remote: bool, background: bool },
UnifiedBackground,
} }
fn write_curated_metrics_plugin(codex_home: &Path) -> Result<PathBuf> { fn write_curated_metrics_plugin(codex_home: &Path) -> Result<PathBuf> {
@@ -324,9 +323,12 @@ async fn assert_plugin_measurement_analytics(runtime: PluginMetricsRuntime) -> R
skip_if_wine_exec!(Ok(()), "plugin metrics fixture is Unix-only"); skip_if_wine_exec!(Ok(()), "plugin metrics fixture is Unix-only");
let codex_home = TempDir::new()?; let codex_home = TempDir::new()?;
let script_path = write_curated_metrics_plugin(codex_home.path())?; let script_path = write_curated_metrics_plugin(codex_home.path())?.canonicalize()?;
let background = matches!(runtime, PluginMetricsRuntime::UnifiedBackground); let (remote, background) = match runtime {
let unified_exec = !matches!(runtime, PluginMetricsRuntime::Classic); PluginMetricsRuntime::Classic => (false, false),
PluginMetricsRuntime::Unified { remote, background } => (remote, background),
};
let unified_exec = matches!(runtime, PluginMetricsRuntime::Unified { .. });
let mut command = vec![ let mut command = vec![
"/bin/sh".to_string(), "/bin/sh".to_string(),
script_path.to_string_lossy().into_owned(), script_path.to_string_lossy().into_owned(),
@@ -339,7 +341,7 @@ async fn assert_plugin_measurement_analytics(runtime: PluginMetricsRuntime) -> R
PluginMetricsRuntime::Classic => { PluginMetricsRuntime::Classic => {
create_shell_command_sse_response(command, /*workdir*/ None, Some(5_000), call_id)? 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!({ let arguments = serde_json::to_string(&json!({
"cmd": shlex::try_join(command.iter().map(String::as_str))?, "cmd": shlex::try_join(command.iter().map(String::as_str))?,
"yield_time_ms": if background { 10 } else { 1_000 }, "yield_time_ms": if background { 10 } else { 1_000 },
@@ -381,11 +383,19 @@ enabled = true
)?; )?;
mount_analytics_capture(&analytics_server, codex_home.path()).await?; 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()) .with_codex_home(codex_home.path())
.without_managed_config() .without_managed_config();
.build() if remote {
.await?; 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??; timeout(Duration::from_secs(10), mcp.initialize()).await??;
let thread_request = mcp let thread_request = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams { .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")] #[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_plugin_script_emits_measurement_analytics() -> Result<()> { 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")] #[cfg_attr(windows, ignore = "plugin metrics fixture is Unix-only")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn unified_background_plugin_script_emits_measurements_after_turn_completion() -> Result<()> { 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
} }
+1
View File
@@ -38,6 +38,7 @@ codex-utils-plugins = { workspace = true }
chrono = { workspace = true } chrono = { workspace = true }
dirs = { workspace = true } dirs = { workspace = true }
flate2 = { workspace = true } flate2 = { workspace = true }
futures = { workspace = true }
http = { workspace = true } http = { workspace = true }
regex = { workspace = true } regex = { workspace = true }
semver = { workspace = true } semver = { workspace = true }
@@ -1,8 +1,15 @@
use crate::ResolvedPluginMetricsOperation; use crate::ResolvedPluginMetricsOperation;
use codex_analytics::PluginMeasurementRow; 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::AdditionalPermissionProfile;
use codex_protocol::models::FileSystemPermissions; use codex_protocol::models::FileSystemPermissions;
use codex_utils_absolute_path::AbsolutePathBuf; 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 serde::Deserialize;
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::collections::BTreeSet; use std::collections::BTreeSet;
@@ -10,6 +17,7 @@ use std::collections::HashMap;
use std::io::Read; use std::io::Read;
use std::io::Seek; use std::io::Seek;
use std::io::SeekFrom; use std::io::SeekFrom;
use std::sync::Arc;
use tempfile::NamedTempFile; use tempfile::NamedTempFile;
use uuid::Uuid; use uuid::Uuid;
@@ -26,14 +34,51 @@ pub struct PluginMeasurementBatch {
} }
pub struct PluginMetricsSidecar { pub struct PluginMetricsSidecar {
output_file: NamedTempFile, output: PluginMetricsOutput,
_output_dir: tempfile::TempDir,
absolute_output_dir: AbsolutePathBuf, absolute_output_dir: AbsolutePathBuf,
output_env_value: String, output_env_value: String,
resolved: ResolvedPluginMetricsOperation, resolved: ResolvedPluginMetricsOperation,
execution_id: String, execution_id: String,
} }
enum PluginMetricsOutput {
Local {
file: NamedTempFile,
_directory: tempfile::TempDir,
},
Remote {
file_stream: tokio::sync::Mutex<FileSystemReadStream>,
_directory: RemotePluginMetricsDirectory,
},
}
struct RemotePluginMetricsDirectory {
filesystem: Arc<dyn ExecutorFileSystem>,
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)] #[derive(Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
struct OutputEnvelope { struct OutputEnvelope {
@@ -65,8 +110,10 @@ impl PluginMetricsSidecar {
let absolute_output_path = AbsolutePathBuf::from_absolute_path(output_file.path()).ok()?; 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(); let output_env_value = absolute_output_path.as_path().to_str()?.to_string();
Some(Self { Some(Self {
output_file, output: PluginMetricsOutput::Local {
_output_dir: sidecar_dir, file: output_file,
_directory: sidecar_dir,
},
absolute_output_dir, absolute_output_dir,
output_env_value, output_env_value,
resolved, resolved,
@@ -74,6 +121,56 @@ impl PluginMetricsSidecar {
}) })
} }
pub async fn create_remote(
environment: &Environment,
resolved: ResolvedPluginMetricsOperation,
) -> Option<Self> {
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<String, String>) { pub fn install_output_env(&self, env: &mut HashMap<String, String>) {
env.insert( env.insert(
PLUGIN_METRICS_OUTPUT_ENV_VAR.to_string(), PLUGIN_METRICS_OUTPUT_ENV_VAR.to_string(),
@@ -83,7 +180,7 @@ impl PluginMetricsSidecar {
#[cfg(test)] #[cfg(test)]
fn absolute_output_path(&self) -> AbsolutePathBuf { 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 { pub fn additional_permissions(&self) -> AdditionalPermissionProfile {
@@ -96,11 +193,32 @@ impl PluginMetricsSidecar {
} }
} }
pub fn finish(mut self, exit_code: i32) -> Option<PluginMeasurementBatch> { pub async fn finish(mut self, exit_code: i32) -> Option<PluginMeasurementBatch> {
if exit_code != 0 { if exit_code != 0 {
return None; 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 { (!rows.is_empty()).then(|| PluginMeasurementBatch {
plugin_id: self.resolved.plugin_id.as_key(), plugin_id: self.resolved.plugin_id.as_key(),
execution_id: self.execution_id, execution_id: self.execution_id,
@@ -119,19 +237,13 @@ pub fn strip_output_env(env: &mut HashMap<String, String>) {
} }
fn parse_output( fn parse_output(
output_file: &mut std::fs::File, contents: &[u8],
resolved: &ResolvedPluginMetricsOperation, resolved: &ResolvedPluginMetricsOperation,
) -> Option<Vec<PluginMeasurementRow>> { ) -> Option<Vec<PluginMeasurementRow>> {
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 { if contents.len() as u64 > MAX_OUTPUT_BYTES {
return None; 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 { if output.version != 1 || output.measurements.len() > MAX_OUTPUT_ROWS {
return None; return None;
} }
@@ -63,8 +63,8 @@ fn resolved_operation() -> ResolvedPluginMetricsOperation {
} }
} }
#[test] #[tokio::test]
fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() { async fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() {
let sidecar = create_sidecar(); let sidecar = create_sidecar();
let path = sidecar.absolute_output_path(); let path = sidecar.absolute_output_path();
std::fs::write( std::fs::write(
@@ -87,7 +87,10 @@ fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() {
) )
.expect("write output"); .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(); let execution_id = batch.execution_id.clone();
assert_eq!( assert_eq!(
batch, batch,
@@ -118,8 +121,8 @@ fn sidecar_keeps_valid_rows_and_first_duplicate_then_cleans_up() {
assert!(!path.exists()); assert!(!path.exists());
} }
#[test] #[tokio::test]
fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() { async fn malformed_oversized_and_nonzero_outputs_are_ignored_and_cleaned_up() {
for output in [ for output in [
r#"{"version":2,"measurements":[]}"#.as_bytes().to_vec(), r#"{"version":2,"measurements":[]}"#.as_bytes().to_vec(),
r#"{"version":1,"measurements":[],"unknown":true}"#.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 sidecar = create_sidecar();
let path = sidecar.absolute_output_path(); let path = sidecar.absolute_output_path();
std::fs::write(path.as_path(), output).expect("write output"); 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()); 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}]}"#, r#"{"version":1,"measurements":[{"name":"files_scanned","value":1}]}"#,
) )
.expect("write output"); .expect("write output");
assert_eq!(sidecar.finish(/*exit_code*/ 1), None); assert_eq!(sidecar.finish(/*exit_code*/ 1).await, None);
assert!(!path.exists()); assert!(!path.exists());
} }
@@ -176,8 +179,8 @@ fn reserved_output_env_is_absent_without_sidecar_and_cannot_be_overridden() {
} }
#[cfg(unix)] #[cfg(unix)]
#[test] #[tokio::test]
fn sidecar_reads_the_original_file_after_path_replacement() { async fn sidecar_reads_the_original_file_after_path_replacement() {
let sidecar = create_sidecar(); let sidecar = create_sidecar();
let path = sidecar.absolute_output_path(); let path = sidecar.absolute_output_path();
std::fs::remove_file(path.as_path()).expect("remove original 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"); .expect("write replacement output");
assert_eq!(sidecar.finish(/*exit_code*/ 0), None); assert_eq!(sidecar.finish(/*exit_code*/ 0).await, None);
assert!(!path.exists()); assert!(!path.exists());
} }
@@ -208,6 +208,13 @@ impl TrustedPluginRoots {
cwd: &AbsolutePathBuf, cwd: &AbsolutePathBuf,
) -> Option<ResolvedPluginMetricsOperation> { ) -> Option<ResolvedPluginMetricsOperation> {
let attribution = self.resolve_attribution(command, cwd)?; let attribution = self.resolve_attribution(command, cwd)?;
self.metrics_operation_for_attribution(attribution)
}
fn metrics_operation_for_attribution(
&self,
attribution: PluginCommandAttribution,
) -> Option<ResolvedPluginMetricsOperation> {
let mut matches = self.roots.iter().filter_map(|root| { let mut matches = self.roots.iter().filter_map(|root| {
(root.plugin_id == attribution.plugin_id) (root.plugin_id == attribution.plugin_id)
.then(|| { .then(|| {
@@ -262,6 +269,19 @@ impl TrustedPluginRoots {
(contents == candidate.contents).then_some(candidate.attribution) (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<ResolvedPluginMetricsOperation> {
let attribution = self
.resolve_executor_attribution(command, cwd, file_system)
.await?;
self.metrics_operation_for_attribution(attribution)
}
fn local_candidate_for_executor_script( fn local_candidate_for_executor_script(
&self, &self,
script: &PathUri, script: &PathUri,
+27 -2
View File
@@ -1,10 +1,35 @@
use crate::session::session::Session; use crate::session::session::Session;
use crate::session::turn_context::TurnContext; use crate::session::turn_context::TurnContext;
use crate::tools::sandboxing::ToolCtx;
use codex_analytics::PluginMeasurementsInput; use codex_analytics::PluginMeasurementsInput;
use codex_core_plugins::PluginMetricsSidecar; 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<PluginMetricsSidecar> {
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. /// 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<PluginMetricsSidecar>, metrics_sidecar: Option<PluginMetricsSidecar>,
exit_code: i32, exit_code: i32,
session: &Session, session: &Session,
@@ -14,7 +39,7 @@ pub(crate) fn finish_and_track_measurements(
let Some(metrics_sidecar) = metrics_sidecar else { let Some(metrics_sidecar) = metrics_sidecar else {
return; return;
}; };
let Some(batch) = metrics_sidecar.finish(exit_code) else { let Some(batch) = metrics_sidecar.finish(exit_code).await else {
return; return;
}; };
session session
+15 -5
View File
@@ -243,14 +243,24 @@ impl TurnContext {
} }
} }
pub(crate) fn plugin_metrics_operation_for_command( pub(crate) async fn plugin_metrics_operation_for_command(
&self, &self,
command: &[String], command: &[String],
cwd: &AbsolutePathBuf, cwd: &PathUri,
environment: &Environment,
) -> Option<ResolvedPluginMetricsOperation> { ) -> Option<ResolvedPluginMetricsOperation> {
self.extension_data let trusted_roots = self.extension_data.get::<TrustedPluginRoots>()?;
.get::<TrustedPluginRoots>()? if environment.is_remote() {
.resolve_metrics_operation(command, cwd) 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 { pub(crate) fn permission_profile(&self) -> PermissionProfile {
+11 -10
View File
@@ -11,6 +11,7 @@ pub(crate) mod zsh_fork_backend;
use crate::exec::ExecCapturePolicy; use crate::exec::ExecCapturePolicy;
use crate::guardian::GuardianNetworkAccessTrigger; use crate::guardian::GuardianNetworkAccessTrigger;
use crate::plugins::metrics::finish_and_track_measurements; use crate::plugins::metrics::finish_and_track_measurements;
use crate::plugins::metrics::sidecar_for_command;
use crate::sandboxing::ExecOptions; use crate::sandboxing::ExecOptions;
use crate::sandboxing::SandboxPermissions; use crate::sandboxing::SandboxPermissions;
use crate::sandboxing::execute_env; use crate::sandboxing::execute_env;
@@ -212,15 +213,14 @@ impl ToolRuntime<ShellRequest, ExecToolCallOutput> for ShellRuntime {
managed_network_for_sandbox_permissions(req.network.as_ref(), sandbox_permissions); managed_network_for_sandbox_permissions(req.network.as_ref(), sandbox_permissions);
let mut env = exec_env_for_sandbox_permissions(&req.env, sandbox_permissions); let mut env = exec_env_for_sandbox_permissions(&req.env, sandbox_permissions);
let explicit_env_overrides = req.explicit_env_overrides.clone(); let explicit_env_overrides = req.explicit_env_overrides.clone();
let metrics_sidecar = (!req.turn_environment.environment.is_remote() let cwd = PathUri::from_abs_path(&req.cwd);
&& ctx.session.services.analytics_events_client.is_enabled()) let metrics_sidecar = sidecar_for_command(
.then(|| { ctx,
ctx.step_context &req.command,
.turn &cwd,
.plugin_metrics_operation_for_command(&req.command, &req.cwd) req.turn_environment.environment.as_ref(),
}) )
.flatten() .await;
.and_then(PluginMetricsSidecar::create);
if let Some(sidecar) = metrics_sidecar.as_ref() { if let Some(sidecar) = metrics_sidecar.as_ref() {
sidecar.install_output_env(&mut env); sidecar.install_output_env(&mut env);
} }
@@ -320,7 +320,8 @@ impl ToolRuntime<ShellRequest, ExecToolCallOutput> for ShellRuntime {
&ctx.session, &ctx.session,
&ctx.step_context.turn, &ctx.step_context.turn,
&ctx.call_id, &ctx.call_id,
); )
.await;
Ok(out) Ok(out)
} }
} }
@@ -9,6 +9,7 @@ use crate::exec::ExecExpiration;
use crate::guardian::GUARDIAN_REVIEW_TIMEOUT; use crate::guardian::GUARDIAN_REVIEW_TIMEOUT;
use crate::guardian::GuardianNetworkAccessTrigger; use crate::guardian::GuardianNetworkAccessTrigger;
use crate::guardian::routes_approval_to_guardian; use crate::guardian::routes_approval_to_guardian;
use crate::plugins::metrics::sidecar_for_command;
use crate::sandboxing::ExecOptions; use crate::sandboxing::ExecOptions;
use crate::sandboxing::ExecServerEnvConfig; use crate::sandboxing::ExecServerEnvConfig;
use crate::sandboxing::SandboxPermissions; use crate::sandboxing::SandboxPermissions;
@@ -337,16 +338,13 @@ impl<'a> ToolRuntime<UnifiedExecRequest, UnifiedExecAttempt> for UnifiedExecRunt
None => (env, None, None), None => (env, None, None),
}; };
let explicit_env_overrides = req.explicit_env_overrides.clone(); let explicit_env_overrides = req.explicit_env_overrides.clone();
let metrics_sidecar = (!environment_is_remote let metrics_sidecar = sidecar_for_command(
&& ctx.session.services.analytics_events_client.is_enabled()) ctx,
.then(|| { &req.command,
let cwd = req.cwd.to_abs_path().ok()?; &req.cwd,
ctx.step_context req.turn_environment.environment.as_ref(),
.turn )
.plugin_metrics_operation_for_command(&req.command, &cwd) .await;
})
.flatten()
.and_then(PluginMetricsSidecar::create);
if let Some(sidecar) = metrics_sidecar.as_ref() { if let Some(sidecar) = metrics_sidecar.as_ref() {
sidecar.install_output_env(&mut env); sidecar.install_output_env(&mut env);
} }
@@ -216,7 +216,8 @@ pub(crate) fn spawn_exit_watcher(
&session_ref, &session_ref,
&turn_ref, &turn_ref,
&call_id, &call_id,
); )
.await;
emit_exec_end_for_unified_exec( emit_exec_end_for_unified_exec(
session_ref, session_ref,
turn_ref, turn_ref,
@@ -251,14 +251,15 @@ struct InitialExecCommandGuard {
} }
impl 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( finish_and_track_measurements(
self.metrics_sidecar.take(), self.metrics_sidecar.take(),
exit_code, exit_code,
&context.session, &context.session,
&context.step_context.turn, &context.step_context.turn,
&context.call_id, &context.call_id,
); )
.await;
} }
} }
@@ -661,7 +662,8 @@ impl UnifiedExecProcessManager {
&context.session, &context.session,
&context.step_context.turn, &context.step_context.turn,
&context.call_id, &context.call_id,
); )
.await;
(None, exit_code) (None, exit_code)
} }
ProcessStatus::Unknown => { ProcessStatus::Unknown => {
@@ -694,7 +696,9 @@ impl UnifiedExecProcessManager {
} }
let exit_code = process.exit_code(); let exit_code = process.exit_code();
let exit = exit_code.unwrap_or(-1); 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( emit_exec_end_for_unified_exec(
Arc::clone(&context.session), Arc::clone(&context.session),
Arc::clone(&context.step_context.turn), Arc::clone(&context.step_context.turn),
+22 -13
View File
@@ -97,6 +97,9 @@ pub struct EnvironmentInfo {
/// On Windows, a command's `TEMP` or `TMP` overrides take precedence. /// On Windows, a command's `TEMP` or `TMP` overrides take precedence.
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub temporary_directories: Option<Vec<PathUri>>, pub temporary_directories: Option<Vec<PathUri>>,
/// Executor-native temporary directory for private, child-visible sidecars.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub temp_dir: Option<PathUri>,
/// Optional executor features that clients must gate before sending newer request fields. /// Optional executor features that clients must gate before sending newer request fields.
#[serde(default)] #[serde(default)]
pub capabilities: EnvironmentCapabilities, pub capabilities: EnvironmentCapabilities,
@@ -145,30 +148,33 @@ impl EnvironmentInfo {
} else { } else {
&["TMPDIR"] &["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(); let mut temporary_directories = Vec::new();
for name in temporary_directory_env_vars { for name in temporary_directory_env_vars {
if let Some(path) = std::env::var_os(name) if let Some(path) = std::env::var_os(name)
.filter(|path| !path.is_empty()) .filter(|path| !path.is_empty())
.filter(|path| cfg!(unix) || std::path::Path::new(path).is_absolute()) .filter(|path| cfg!(unix) || std::path::Path::new(path).is_absolute())
.and_then(|path| { .and_then(&normalize_temp_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
}
})
})
&& !temporary_directories.contains(&path) && !temporary_directories.contains(&path)
{ {
temporary_directories.push(path); temporary_directories.push(path);
} }
} }
let temp_dir = normalize_temp_path(std::env::temp_dir().into_os_string());
Self { Self {
shell: codex_shell_command::shell_detect::default_user_shell().into(), shell: codex_shell_command::shell_detect::default_user_shell().into(),
cwd: cwd.and_then(|cwd| PathUri::from_host_native_path(cwd).ok()), cwd: cwd.and_then(|cwd| PathUri::from_host_native_path(cwd).ok()),
temporary_directories: Some(temporary_directories), temporary_directories: Some(temporary_directories),
temp_dir,
capabilities: EnvironmentCapabilities { capabilities: EnvironmentCapabilities {
network_proxy_launch: true, network_proxy_launch: true,
capability_discovery_sandbox: true, capability_discovery_sandbox: true,
@@ -417,6 +423,9 @@ pub struct FsCreateDirectoryParams {
pub path: PathUri, pub path: PathUri,
pub recursive: Option<bool>, pub recursive: Option<bool>,
pub sandbox: Option<FileSystemSandboxContext>, pub sandbox: Option<FileSystemSandboxContext>,
/// Atomically restrict a newly created, non-recursive directory to its owner.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub private: Option<bool>,
} }
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -902,6 +911,7 @@ mod tests {
}, },
cwd: None, cwd: None,
temporary_directories: None, temporary_directories: None,
temp_dir: None,
capabilities: EnvironmentCapabilities::default(), capabilities: EnvironmentCapabilities::default(),
} }
); );
@@ -1000,10 +1010,9 @@ mod tests {
.join("relative-temp"), .join("relative-temp"),
) )
.expect("absolute temporary directory URI"); .expect("absolute temporary directory URI");
assert_eq!( let info = EnvironmentInfo::local();
EnvironmentInfo::local().temporary_directories, assert_eq!(info.temporary_directories, Some(vec![expected.clone()]));
Some(vec![expected]) assert_eq!(info.temp_dir, Some(expected));
);
} }
#[test] #[test]
+26
View File
@@ -37,9 +37,11 @@ use crate::local_file_system::LocalFileSystem;
use crate::local_process::LocalProcess; use crate::local_process::LocalProcess;
use crate::process::ExecBackend; use crate::process::ExecBackend;
use crate::protocol::EnvironmentInfo; use crate::protocol::EnvironmentInfo;
use crate::protocol::FsCreateDirectoryParams;
use crate::remote::NoiseRendezvousEnvironmentConfig; use crate::remote::NoiseRendezvousEnvironmentConfig;
use crate::remote_file_system::RemoteFileSystem; use crate::remote_file_system::RemoteFileSystem;
use crate::remote_process::RemoteProcess; use crate::remote_process::RemoteProcess;
use codex_utils_path_uri::PathUri;
use tokio::sync::watch; use tokio::sync::watch;
use tokio_util::task::AbortOnDropHandle; 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. /// Discovers plugin and skill manifests through the environment's high-level discovery API.
pub async fn discover_capability_roots( pub async fn discover_capability_roots(
&self, &self,
@@ -1078,6 +1100,10 @@ mod tests {
.expect("cwd URI") .expect("cwd URI")
) )
); );
assert_eq!(
info.temp_dir,
PathUri::from_host_native_path(std::env::temp_dir()).ok()
);
} }
#[tokio::test] #[tokio::test]
@@ -144,6 +144,7 @@ impl RemoteFileSystem {
path: path.clone(), path: path.clone(),
recursive: Some(options.recursive), recursive: Some(options.recursive),
sandbox: remote_sandbox_context(sandbox), sandbox: remote_sandbox_context(sandbox),
private: None,
}) })
.await; .await;
self.metadata_requests.lock().await.clear(); self.metadata_requests.lock().await.clear();
@@ -138,6 +138,7 @@ impl SandboxedFileSystem {
path: path.clone(), path: path.clone(),
recursive: Some(options.recursive), recursive: Some(options.recursive),
sandbox: None, sandbox: None,
private: None,
}), }),
) )
.await? .await?
@@ -153,6 +153,27 @@ impl FileSystemHandler {
&self, &self,
params: FsCreateDirectoryParams, params: FsCreateDirectoryParams,
) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> { ) -> Result<FsCreateDirectoryResponse, JSONRPCErrorError> {
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); let recursive = params.recursive.unwrap_or(true);
self.file_system self.file_system
.create_directory( .create_directory(
@@ -289,6 +310,9 @@ fn map_fs_error(err: io::Error) -> JSONRPCErrorError {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
use codex_protocol::protocol::NetworkAccess; use codex_protocol::protocol::NetworkAccess;
use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SandboxPolicy;
use codex_utils_path_uri::PathUri; use codex_utils_path_uri::PathUri;
@@ -299,6 +323,62 @@ mod tests {
use crate::protocol::FsReadFileParams; use crate::protocol::FsReadFileParams;
use crate::protocol::FsWriteFileParams; 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] #[tokio::test]
async fn no_platform_sandbox_policies_do_not_require_configured_sandbox_helper() { async fn no_platform_sandbox_policies_do_not_require_configured_sandbox_helper() {
let temp_dir = tempfile::tempdir().expect("tempdir"); let temp_dir = tempfile::tempdir().expect("tempdir");
+2
View File
@@ -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( expected_environment_info.temporary_directories = Some(vec![PathUri::from_host_native_path(
temporary_directory.path(), temporary_directory.path(),
)?]); )?]);
expected_environment_info.temp_dir =
Some(PathUri::from_host_native_path(temporary_directory.path())?);
assert_eq!( assert_eq!(
serde_json::from_value::<EnvironmentInfo>(result)?, serde_json::from_value::<EnvironmentInfo>(result)?,
expected_environment_info expected_environment_info