From f53c91be2c1393ad5fd69b4d2a95445812598087 Mon Sep 17 00:00:00 2001 From: viyatb-oai Date: Wed, 2 Sep 2026 20:32:12 +0000 Subject: [PATCH] Add attributed exec process lifecycle telemetry (#42373) ## Why Exec-server process and network-policy events need stable attribution to the launching tool call and executor without recording process payloads. ## What changed - Add optional `ExecMetadata` to `ExecParams` and propagate the thread and tool call IDs from unified exec. - Emit bounded OpenTelemetry events for process start, spawn failure, sandbox denial, and exit. Correlate them with the launch trace and registry-issued executor identity while excluding arguments, paths, environment values, output, and error text. - Preserve launch attribution across long-running processes and reconnects, and attach the same metadata to network-policy audit events. - Keep the protocol backward compatible when metadata is omitted, and prevent invalid trace headers from inheriting an unrelated active span. ## Testing Add coverage for metadata serialization and propagation, lifecycle event fields, trace relationships, reconnect behavior, spawn failures, sandbox denials, and network-policy attribution. GitOrigin-RevId: 7aa480a7289c73cb95e2c124c35500bb6f0d5084 --- codex-rs/Cargo.lock | 1 + codex-rs/cli/tests/exec_server.rs | 1 + .../core/src/tools/runtimes/unified_exec.rs | 2 + codex-rs/core/src/unified_exec/mod_tests.rs | 4 + .../core/src/unified_exec/process_manager.rs | 9 + .../src/unified_exec/process_manager_tests.rs | 2 + .../core/src/unified_exec/shell_snapshot.rs | 1 + .../suite/unified_exec_process_events.rs | 7 + codex-rs/exec-server-protocol/src/protocol.rs | 32 + codex-rs/exec-server/Cargo.toml | 2 +- codex-rs/exec-server/src/client.rs | 1 + .../src/client/tests/network_policy_tests.rs | 1 + .../exec-server/src/client_refresh_tests.rs | 1 + codex-rs/exec-server/src/environment.rs | 2 + codex-rs/exec-server/src/forward_tests.rs | 2 + codex-rs/exec-server/src/lib.rs | 2 + codex-rs/exec-server/src/local_process.rs | 250 +++--- .../src/noise_relay/executor_stream.rs | 3 + .../src/noise_relay/executor_stream_tests.rs | 1 + .../src/noise_relay/forward_stream_tests.rs | 1 + .../src/noise_relay/stream_handler.rs | 8 +- codex-rs/exec-server/src/process_sandbox.rs | 20 +- .../exec-server/src/process_sandbox_tests.rs | 34 +- codex-rs/exec-server/src/process_telemetry.rs | 78 ++ codex-rs/exec-server/src/relay.rs | 6 + codex-rs/exec-server/src/relay_noise_tests.rs | 56 +- codex-rs/exec-server/src/rpc.rs | 15 +- codex-rs/exec-server/src/server.rs | 4 + codex-rs/exec-server/src/server/handler.rs | 22 +- .../exec-server/src/server/handler/tests.rs | 38 +- .../exec-server/src/server/process_handler.rs | 9 +- .../src/server/process_otel_tests.rs | 752 ++++++++++++++++++ codex-rs/exec-server/src/server/processor.rs | 56 +- codex-rs/exec-server/src/server/registry.rs | 11 +- .../exec-server/src/shell_snapshot_tests.rs | 10 +- codex-rs/exec-server/src/telemetry.rs | 21 + .../exec-server/tests/accepted_websocket.rs | 1 + codex-rs/exec-server/tests/exec_process.rs | 20 + codex-rs/exec-server/tests/forward.rs | 1 + codex-rs/exec-server/tests/relay.rs | 1 + codex-rs/network-proxy/Cargo.toml | 1 + codex-rs/network-proxy/src/lib.rs | 3 + codex-rs/network-proxy/src/network_policy.rs | 103 ++- .../network-proxy/src/process_log_metadata.rs | 17 + codex-rs/network-proxy/src/runtime.rs | 17 + codex-rs/otel/src/trace_context.rs | 9 +- .../rmcp-client/src/stdio_server_launcher.rs | 1 + 47 files changed, 1461 insertions(+), 178 deletions(-) create mode 100644 codex-rs/exec-server/src/process_telemetry.rs create mode 100644 codex-rs/exec-server/src/server/process_otel_tests.rs create mode 100644 codex-rs/network-proxy/src/process_log_metadata.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 9a31b5f104..ce9b1f2a12 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3916,6 +3916,7 @@ dependencies = [ "codex-utils-rustls-provider", "codex-windows-sandbox", "globset", + "opentelemetry", "pretty_assertions", "rama-core", "rama-http", diff --git a/codex-rs/cli/tests/exec_server.rs b/codex-rs/cli/tests/exec_server.rs index 6e975a6226..0ce5ef814e 100644 --- a/codex-rs/cli/tests/exec_server.rs +++ b/codex-rs/cli/tests/exec_server.rs @@ -268,6 +268,7 @@ metrics_exporter = {{ otlp-http = {{ endpoint = "{collector_url}/v1/metrics", pr .map_err(|()| anyhow::anyhow!("could not convert cwd to file URL"))?; client .exec(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("parent-lifetime-process"), argv: argv.into_iter().map(str::to_string).collect(), cwd: cwd.as_str().parse()?, diff --git a/codex-rs/core/src/tools/runtimes/unified_exec.rs b/codex-rs/core/src/tools/runtimes/unified_exec.rs index aea36d9f48..83926d3f66 100644 --- a/codex-rs/core/src/tools/runtimes/unified_exec.rs +++ b/codex-rs/core/src/tools/runtimes/unified_exec.rs @@ -476,6 +476,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt .open_session_with_prepared_exec_env( req.process_id, &prepared.exec_request, + Some(ctx), windows_sandbox_proxy_settings_mode, /*network_policy_decider*/ None, req.tty, @@ -523,6 +524,7 @@ impl<'a> ToolRuntime for UnifiedExecRunt .manager .open_session_with_exec_env( req.process_id, + ctx, command, options, attempt, diff --git a/codex-rs/core/src/unified_exec/mod_tests.rs b/codex-rs/core/src/unified_exec/mod_tests.rs index dc32b21ee1..d090847a15 100644 --- a/codex-rs/core/src/unified_exec/mod_tests.rs +++ b/codex-rs/core/src/unified_exec/mod_tests.rs @@ -111,6 +111,7 @@ async fn exec_command_with_tty( .open_session_with_prepared_exec_env( process_id, &request, + /*tool_ctx*/ None, codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, /*network_policy_decider*/ None, tty, @@ -772,6 +773,7 @@ async fn completed_pipe_commands_preserve_exit_code() -> anyhow::Result<()> { .open_session_with_prepared_exec_env( /*process_id*/ 1234, &request, + /*tool_ctx*/ None, codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, /*network_policy_decider*/ None, /*tty*/ false, @@ -814,6 +816,7 @@ async fn unified_exec_uses_remote_exec_server_when_configured() -> anyhow::Resul .open_session_with_prepared_exec_env( /*process_id*/ 1234, &request, + /*tool_ctx*/ None, codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, /*network_policy_decider*/ None, /*tty*/ true, @@ -863,6 +866,7 @@ async fn remote_exec_server_rejects_inherited_fd_launches() -> anyhow::Result<() .open_session_with_prepared_exec_env( /*process_id*/ 1234, &request, + /*tool_ctx*/ None, codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, /*network_policy_decider*/ None, /*tty*/ true, diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 21218570af..81d91c73a4 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -213,6 +213,7 @@ fn exec_server_env_for_request( fn exec_server_params_for_request( process_id: i32, request: &ExecRequest, + tool_ctx: Option<&ToolCtx>, windows_sandbox_proxy_settings_mode: codex_sandboxing::WindowsSandboxProxySettingsMode, tty: bool, ) -> codex_exec_server::ExecParams { @@ -231,6 +232,10 @@ fn exec_server_params_for_request( }; codex_exec_server::ExecParams { process_id: exec_server_process_id.into(), + metadata: tool_ctx.map(|ctx| codex_exec_server::ExecMetadata { + thread_id: Some(ctx.session.thread_id()), + tool_call_id: Some(ctx.call_id.clone()), + }), argv: request.command.clone(), cwd: request.cwd.clone(), env_policy, @@ -1176,6 +1181,7 @@ impl UnifiedExecProcessManager { pub(crate) async fn open_session_with_exec_env( &self, process_id: i32, + tool_ctx: &ToolCtx, command: SandboxCommand, options: ExecOptions, attempt: &SandboxAttempt<'_>, @@ -1208,6 +1214,7 @@ impl UnifiedExecProcessManager { self.open_session_with_prepared_exec_env( process_id, &request, + Some(tool_ctx), windows_sandbox_proxy_settings_mode, network_policy_decider, tty, @@ -1231,6 +1238,7 @@ impl UnifiedExecProcessManager { &self, process_id: i32, request: &ExecRequest, + tool_ctx: Option<&ToolCtx>, windows_sandbox_proxy_settings_mode: codex_sandboxing::WindowsSandboxProxySettingsMode, network_policy_decider: Option>, tty: bool, @@ -1250,6 +1258,7 @@ impl UnifiedExecProcessManager { let params = exec_server_params_for_request( process_id, request, + tool_ctx, windows_sandbox_proxy_settings_mode, tty, ); diff --git a/codex-rs/core/src/unified_exec/process_manager_tests.rs b/codex-rs/core/src/unified_exec/process_manager_tests.rs index 523e7eb513..c1103e569b 100644 --- a/codex-rs/core/src/unified_exec/process_manager_tests.rs +++ b/codex-rs/core/src/unified_exec/process_manager_tests.rs @@ -218,6 +218,7 @@ fn exec_server_params_use_path_uri_and_env_policy_overlay_contract() { exec_server_params_for_request( /*process_id*/ 123, request, + /*tool_ctx*/ None, proxy_settings_mode, /*tty*/ true, ) @@ -225,6 +226,7 @@ fn exec_server_params_use_path_uri_and_env_policy_overlay_contract() { let params = params_for_request(&request); assert_eq!(params.process_id.as_str(), "123"); + assert_eq!(params.metadata, None); assert_eq!(params.cwd, request.cwd); assert!(params.enforce_managed_network); assert_eq!(params.managed_network, Some(managed_network)); diff --git a/codex-rs/core/src/unified_exec/shell_snapshot.rs b/codex-rs/core/src/unified_exec/shell_snapshot.rs index 43e95cc1cb..fdce56e3b3 100644 --- a/codex-rs/core/src/unified_exec/shell_snapshot.rs +++ b/codex-rs/core/src/unified_exec/shell_snapshot.rs @@ -130,6 +130,7 @@ impl Session { } let params = ExecParams { process_id: format!("shell-snapshot-warmup-{}", Uuid::new_v4()).into(), + metadata: None, // This capture-only operation never starts the command or routes // speculative work through tool approvals or exec-policy bypass. argv: shell.derive_exec_args("", /*use_login_shell*/ true), diff --git a/codex-rs/core/tests/suite/unified_exec_process_events.rs b/codex-rs/core/tests/suite/unified_exec_process_events.rs index 8a529ab34e..75e0b16e10 100644 --- a/codex-rs/core/tests/suite/unified_exec_process_events.rs +++ b/codex-rs/core/tests/suite/unified_exec_process_events.rs @@ -969,6 +969,13 @@ timeout = 900 let exec_server_result = timeout(cleanup_timeout, exec_server) .await .context("fake exec-server should observe process cleanup")??; + assert_eq!( + exec_server_result.process_start["params"]["metadata"], + json!({ + "threadId": test.session_configured.thread_id, + "toolCallId": CALL_ID, + }), + ); if foreign_cwd { let params = &exec_server_result.process_start["params"]; assert_eq!(params["cwd"], "file:///C:/workspace"); diff --git a/codex-rs/exec-server-protocol/src/protocol.rs b/codex-rs/exec-server-protocol/src/protocol.rs index 16150d3a60..2b95ea64f8 100644 --- a/codex-rs/exec-server-protocol/src/protocol.rs +++ b/codex-rs/exec-server-protocol/src/protocol.rs @@ -7,6 +7,7 @@ pub use codex_file_system::WalkOptions; pub use codex_file_system::WalkOutcome; use codex_network_proxy::ManagedNetworkSandboxContext; use codex_network_proxy::RemoteNetworkProxyLaunchConfig; +use codex_protocol::ThreadId; use codex_protocol::capabilities::SelectedCapabilityRoot; use codex_protocol::config_types::ShellEnvironmentPolicyInherit; use codex_shell_command::shell_detect::DetectedShell; @@ -260,12 +261,25 @@ impl From for ShellInfo { } } +/// Optional tool attribution for executor telemetry, not authorization. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ExecMetadata { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thread_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ExecParams { /// Client-chosen logical process handle scoped to this connection/session. /// This is a protocol key, not an OS pid. pub process_id: ProcessId, + /// Optional attribution; older clients omit it and older executors ignore it. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option, pub argv: Vec, /// Working directory URI, interpreted using the exec-server host's path rules at launch time. pub cwd: PathUri, @@ -881,6 +895,7 @@ mod tests { use super::EnvironmentCapabilities; use super::EnvironmentInfo; use super::ExecExitedNotification; + use super::ExecMetadata; use super::ExecParams; use super::ExecResponse; use super::FsReadFileParams; @@ -914,6 +929,10 @@ mod tests { .expect("cwd URI"); let params = ExecParams { process_id: ProcessId::from("managed-network"), + metadata: Some(ExecMetadata { + thread_id: Some(codex_protocol::ThreadId::new()), + tool_call_id: Some("call-1".to_string()), + }), argv: vec!["true".to_string()], cwd, env_policy: None, @@ -942,6 +961,14 @@ mod tests { }; let mut serialized = serde_json::to_value(¶ms).expect("serialize exec params"); + assert_eq!( + ( + serialized.get("threadId").cloned(), + serialized.get("toolCallId").cloned(), + serialized.get("metadata").cloned(), + ), + (None, None, Some(serde_json::json!(params.metadata)),) + ); assert_eq!( serialized["managedNetwork"], serde_json::json!({ @@ -965,14 +992,19 @@ mod tests { .as_object_mut() .expect("exec params object") .remove("networkProxy"); + serialized.as_object_mut().unwrap().remove("metadata"); let legacy: ExecParams = serde_json::from_value(serialized).expect("deserialize legacy exec params"); assert!(legacy.enforce_managed_network); assert_eq!(legacy.managed_network, None); assert_eq!(legacy.network_proxy, None); + assert_eq!(legacy.metadata, None); let legacy_serialized = serde_json::to_value(&legacy).expect("serialize exec params without proxy launch"); assert!(legacy_serialized.get("networkProxy").is_none()); + assert!(legacy_serialized.get("threadId").is_none()); + assert!(legacy_serialized.get("toolCallId").is_none()); + assert!(legacy_serialized.get("metadata").is_none()); } #[test] diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index ffad2503f8..20053588e1 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/codex-rs/exec-server/Cargo.toml @@ -36,6 +36,7 @@ codex-websocket-client = { workspace = true } dirs = { workspace = true } futures = { workspace = true } http = { workspace = true } +opentelemetry = { workspace = true } prost = "0.14.3" serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } @@ -79,7 +80,6 @@ codex-exec-server-test-support = { workspace = true } codex-test-binary-support = { workspace = true } ctor = { workspace = true } http = { workspace = true } -opentelemetry = { workspace = true } opentelemetry_sdk = { workspace = true } pretty_assertions = { workspace = true } rcgen = { workspace = true } diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 94eecc2c0e..eb76b6f373 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -1871,6 +1871,7 @@ mod tests { let session = client .start_process( ExecParams { + metadata: Default::default(), process_id: process_id.clone(), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) diff --git a/codex-rs/exec-server/src/client/tests/network_policy_tests.rs b/codex-rs/exec-server/src/client/tests/network_policy_tests.rs index 468c3dfc7a..5af5acf97f 100644 --- a/codex-rs/exec-server/src/client/tests/network_policy_tests.rs +++ b/codex-rs/exec-server/src/client/tests/network_policy_tests.rs @@ -266,6 +266,7 @@ async fn abandoned_process_start_unregisters_and_cleans_up() { let start_process_id = process_id.clone(); let start = tokio::spawn(async move { let params = ExecParams { + metadata: Default::default(), process_id: start_process_id, argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) diff --git a/codex-rs/exec-server/src/client_refresh_tests.rs b/codex-rs/exec-server/src/client_refresh_tests.rs index 34dab7bb14..4e45a85ec5 100644 --- a/codex-rs/exec-server/src/client_refresh_tests.rs +++ b/codex-rs/exec-server/src/client_refresh_tests.rs @@ -561,6 +561,7 @@ async fn retirement_rejects_pending_process_start_before_stream_cleanup() -> Res let process_id = ProcessId::from("retired-process"); let call = rpc.client.start_process( crate::protocol::ExecParams { + metadata: Default::default(), process_id: process_id.clone(), argv: vec!["unused".to_owned()], cwd: "file:///".parse()?, diff --git a/codex-rs/exec-server/src/environment.rs b/codex-rs/exec-server/src/environment.rs index 953f8d4da2..0cb6df392a 100644 --- a/codex-rs/exec-server/src/environment.rs +++ b/codex-rs/exec-server/src/environment.rs @@ -1797,6 +1797,7 @@ mod tests { let response = environment .get_exec_backend() .start(crate::ExecParams { + metadata: Default::default(), process_id: ProcessId::from("default-env-proc"), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path( @@ -1839,6 +1840,7 @@ mod tests { let result = environment .get_exec_backend() .start(crate::ExecParams { + metadata: Default::default(), process_id: ProcessId::from("local-sandbox-proc"), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path( diff --git a/codex-rs/exec-server/src/forward_tests.rs b/codex-rs/exec-server/src/forward_tests.rs index a6740ca882..069e5d764c 100644 --- a/codex-rs/exec-server/src/forward_tests.rs +++ b/codex-rs/exec-server/src/forward_tests.rs @@ -38,6 +38,7 @@ async fn transport_disconnect_cancels_an_unfinished_websocket_handshake() -> Res outgoing_tx, disconnected_rx, writer_task: tokio::spawn(async {}), + executor_registration: None, }; let task = tokio::spawn(forwarder.run_connection(connection)); let deadline = Duration::from_secs(5); @@ -64,6 +65,7 @@ async fn destination_close_is_acknowledged() -> Result<()> { outgoing_tx, disconnected_rx, writer_task: tokio::spawn(async {}), + executor_registration: None, })); let deadline = Duration::from_secs(5); let (socket, _) = timeout(deadline, listener.accept()).await??; diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index f0e23038bb..aa61d220f2 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -24,6 +24,7 @@ mod noise_channel; mod noise_relay; mod process; mod process_sandbox; +mod process_telemetry; mod regular_file; mod relay; mod relay_proto; @@ -142,6 +143,7 @@ pub use protocol::EnvironmentStatusKind; pub use protocol::ExecClosedNotification; pub use protocol::ExecEnvPolicy; pub use protocol::ExecExitedNotification; +pub use protocol::ExecMetadata; pub use protocol::ExecOutputDeltaNotification; pub use protocol::ExecOutputStream; pub use protocol::ExecParams; diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index f45e7d472a..c728c7416e 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -7,6 +7,8 @@ use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering; use std::time::Duration; +use crate::process_telemetry::ProcessTelemetry; +use crate::process_telemetry::ProcessTelemetryEvent; use codex_exec_server_protocol::JSONRPCErrorError; use codex_network_proxy::NetworkPolicyAuditEvent; use codex_network_proxy::NetworkPolicyAuditObserver; @@ -21,11 +23,15 @@ use codex_sandboxing::SandboxType; use codex_sandboxing::is_likely_sandbox_denied; use codex_utils_pty::ExecCommandSession; use codex_utils_pty::ProcessSignal as PtyProcessSignal; +use opentelemetry::trace::SpanContext; +use opentelemetry::trace::TraceContextExt; use tokio::sync::Mutex; use tokio::sync::Notify; use tokio::sync::mpsc; use tokio::sync::watch; use tokio_util::sync::CancellationToken; +use tracing::Instrument; +use tracing::instrument::WithSubscriber; use crate::ExecBackend; use crate::ExecBackendFuture; @@ -40,7 +46,7 @@ use crate::StartedExecProcess; use crate::network_policy_decisions::network_policy_decider; use crate::process::ExecProcessEventLog; use crate::process::sandbox_type_from_protocol; -use crate::process_sandbox::prepare_exec_request; +use crate::process_sandbox::prepare_exec_request_with_telemetry; use crate::protocol::EXEC_CLOSED_METHOD; use crate::protocol::ExecClosedNotification; use crate::protocol::ExecEnvPolicy; @@ -275,7 +281,24 @@ impl LocalProcess { async fn start_process( &self, params: ExecParams, + mut telemetry: ProcessTelemetry, ) -> Result<(ExecResponse, watch::Sender, ExecProcessEventLog), JSONRPCErrorError> { + telemetry.launch_context = telemetry.launch_context.filter(SpanContext::is_valid); + let metadata = params.metadata.as_ref(); + telemetry.thread_id = metadata + .and_then(|metadata| metadata.thread_id.as_ref()) + .map(ToString::to_string); + // Correlation is controller-supplied, not authorization or arbitrary diagnostic text. + telemetry.tool_call_id = metadata + .and_then(|metadata| metadata.tool_call_id.as_ref()) + .filter(|id| { + !id.is_empty() + && id.len() <= 256 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"_-.:".contains(&byte)) + }) + .cloned(); let process_id = params.process_id.clone(); let policy_decision_timeout_ms = params .network_proxy @@ -344,12 +367,13 @@ impl LocalProcess { "shell snapshots are unsupported on this platform".to_string(), )); } - let prepared = prepare_exec_request( + let prepared = prepare_exec_request_with_telemetry( ¶ms, child_env(¶ms), self.runtime_paths.as_ref(), network_policy_decider, network_policy_audit_observer, + &telemetry, ) .await?; #[cfg(unix)] @@ -403,6 +427,7 @@ impl LocalProcess { let spawned = match spawned_result { Ok(spawned) => spawned, Err(err) => { + telemetry.log(ProcessTelemetryEvent::SpawnFailed, prepared.sandbox); let mut process_map = self.inner.processes.lock().await; if matches!( process_map.get(&process_id), @@ -461,6 +486,7 @@ impl LocalProcess { })), ); } + telemetry.log(ProcessTelemetryEvent::Start, prepared.sandbox); tokio::spawn(stream_output( process_id.clone(), if params.tty { @@ -483,12 +509,17 @@ impl LocalProcess { Arc::clone(&self.inner), Arc::clone(&output_notify), )); - tokio::spawn(watch_exit( - process_id.clone(), - spawned.exit_rx, - Arc::clone(&self.inner), - output_notify, - )); + // Keep the subscriber, but let the request span close independently of process completion. + tokio::spawn( + watch_exit( + process_id.clone(), + spawned.exit_rx, + Arc::clone(&self.inner), + output_notify, + telemetry, + ) + .with_current_subscriber(), + ); Ok(( ExecResponse { @@ -500,8 +531,12 @@ impl LocalProcess { )) } - pub(crate) async fn exec(&self, params: ExecParams) -> Result { - self.start_process(params) + pub(crate) async fn exec( + &self, + params: ExecParams, + telemetry: ProcessTelemetry, + ) -> Result { + self.start_process(params, telemetry) .await .map(|(response, _, _)| response) } @@ -740,7 +775,7 @@ pub(crate) fn shell_environment_policy(env_policy: &ExecEnvPolicy) -> ShellEnvir impl LocalProcess { async fn start(&self, params: ExecParams) -> Result { let (response, wake_tx, events) = self - .start_process(params) + .start_process(params, ProcessTelemetry::default()) .await .map_err(map_handler_error)?; let sandbox_type = sandbox_type_from_protocol(response.sandbox_type); @@ -772,12 +807,13 @@ impl ExecBackend for LocalProcess { "shell snapshot prewarming does not support managed networking".to_string(), )); } - let mut prepared = prepare_exec_request( + let mut prepared = prepare_exec_request_with_telemetry( ¶ms, child_env(¶ms), self.runtime_paths.as_ref(), /*network_policy_decider*/ None, /*network_policy_audit_observer*/ None, + &ProcessTelemetry::default(), ) .await .map_err(map_handler_error)?; @@ -987,93 +1023,115 @@ async fn stream_output( finish_output_stream(process_id, inner).await; } -async fn watch_exit( +fn watch_exit( process_id: ProcessId, exit_rx: tokio::sync::oneshot::Receiver, inner: Arc, output_notify: Arc, -) { - let exit_code = exit_rx.await.unwrap_or(-1); - let sandboxed = { - let mut processes = inner.processes.lock().await; - match processes.get_mut(&process_id) { - Some(ProcessEntry::Running(process)) => { - let sandboxed = process.sandbox != SandboxType::None; - if let Some(metrics) = process.metrics.take() { - metrics.finish(if process.termination_requested { - "terminated" - } else if exit_code == 0 { - "success" - } else { - "error" - }); - } - sandboxed - } - Some(ProcessEntry::Starting(_)) | None => false, - } - }; - if sandboxed { - let _ = tokio::time::timeout(Duration::from_millis(20), output_notify.notified()).await; + telemetry: ProcessTelemetry, +) -> impl std::future::Future + Send { + // Set the copied OTEL parent before entering; never retain the RPC tracing span. + let process_span = tracing::info_span!(parent: None, "codex.exec_server.process"); + if let Some(launch_context) = &telemetry.launch_context { + codex_otel::set_parent_from_context( + &process_span, + opentelemetry::Context::new().with_remote_span_context(launch_context.clone()), + ); } - let notification = { - let mut processes = inner.processes.lock().await; - if let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) { - let seq = process.next_seq; - process.next_seq += 1; - process.exit_code = Some(exit_code); - if process.sandbox != SandboxType::None { - let mut stdout = Vec::new(); - let mut stderr = Vec::new(); - let mut aggregated = Vec::new(); - for chunk in &process.output { - match chunk.stream { - ExecOutputStream::Stdout | ExecOutputStream::Pty => { - stdout.extend_from_slice(&chunk.chunk); - } - ExecOutputStream::Stderr => stderr.extend_from_slice(&chunk.chunk), + async move { + let exit_code = exit_rx.await.unwrap_or(-1); + let sandboxed = { + let mut processes = inner.processes.lock().await; + match processes.get_mut(&process_id) { + Some(ProcessEntry::Running(process)) => { + let sandboxed = process.sandbox != SandboxType::None; + if let Some(metrics) = process.metrics.take() { + metrics.finish(if process.termination_requested { + "terminated" + } else if exit_code == 0 { + "success" + } else { + "error" + }); } - aggregated.extend_from_slice(&chunk.chunk); + sandboxed } - let exec_output = ExecToolCallOutput { - exit_code, - stdout: StreamOutput::new(String::from_utf8_lossy(&stdout).into_owned()), - stderr: StreamOutput::new(String::from_utf8_lossy(&stderr).into_owned()), - aggregated_output: StreamOutput::new( - String::from_utf8_lossy(&aggregated).into_owned(), - ), - ..Default::default() - }; - // Transport the classification to the caller; recording there - // attaches audit context once and avoids duplicate events. - process.sandbox_denied = is_likely_sandbox_denied(process.sandbox, &exec_output); + Some(ProcessEntry::Starting(_)) | None => false, } - let _ = process.wake_tx.send(seq); - process.events.publish(ExecProcessEvent::Exited { - seq, - exit_code, - sandbox_denied: Some(process.sandbox_denied), - }); - Some(ExecExitedNotification { - process_id: process_id.clone(), - seq, - exit_code, - sandbox_denied: Some(process.sandbox_denied), - }) - } else { - None + }; + if sandboxed { + let _ = tokio::time::timeout(Duration::from_millis(20), output_notify.notified()).await; + } + let notification = { + let mut processes = inner.processes.lock().await; + if let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) { + let seq = process.next_seq; + process.next_seq += 1; + process.exit_code = Some(exit_code); + if process.sandbox != SandboxType::None { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + let mut aggregated = Vec::new(); + for chunk in &process.output { + match chunk.stream { + ExecOutputStream::Stdout | ExecOutputStream::Pty => { + stdout.extend_from_slice(&chunk.chunk); + } + ExecOutputStream::Stderr => stderr.extend_from_slice(&chunk.chunk), + } + aggregated.extend_from_slice(&chunk.chunk); + } + let exec_output = ExecToolCallOutput { + exit_code, + stdout: StreamOutput::new(String::from_utf8_lossy(&stdout).into_owned()), + stderr: StreamOutput::new(String::from_utf8_lossy(&stderr).into_owned()), + aggregated_output: StreamOutput::new( + String::from_utf8_lossy(&aggregated).into_owned(), + ), + ..Default::default() + }; + // Keep the classification in the result for caller approval/retry handling. + process.sandbox_denied = + is_likely_sandbox_denied(process.sandbox, &exec_output); + if process.sandbox_denied { + telemetry.log(ProcessTelemetryEvent::SandboxDenied, process.sandbox); + } + } + telemetry.log( + ProcessTelemetryEvent::Exit { + exit_code, + termination_requested: process.termination_requested, + }, + process.sandbox, + ); + let _ = process.wake_tx.send(seq); + process.events.publish(ExecProcessEvent::Exited { + seq, + exit_code, + sandbox_denied: Some(process.sandbox_denied), + }); + Some(ExecExitedNotification { + process_id: process_id.clone(), + seq, + exit_code, + sandbox_denied: Some(process.sandbox_denied), + }) + } else { + None + } + }; + output_notify.notify_waiters(); + if let Some(notification) = notification + && let Some(notifications) = notification_sender(&inner) + { + let _ = notifications + .notify(crate::protocol::EXEC_EXITED_METHOD, ¬ification) + .await; } - }; - output_notify.notify_waiters(); - if let Some(notification) = notification - && let Some(notifications) = notification_sender(&inner) - { - let _ = notifications - .notify(crate::protocol::EXEC_EXITED_METHOD, ¬ification) - .await; - } - maybe_emit_closed(process_id, Arc::clone(&inner)).await; + maybe_emit_closed(process_id, Arc::clone(&inner)).await; + } + .instrument(process_span) } async fn finish_output_stream(process_id: ProcessId, inner: Arc) { @@ -1194,6 +1252,7 @@ mod tests { fn test_exec_params(env: HashMap) -> ExecParams { ExecParams { + metadata: None, process_id: ProcessId::from("env-test"), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) @@ -1237,7 +1296,7 @@ mod tests { .for_execution("environment-1".to_string(), "execution-1".to_string()), ); backend - .exec(params) + .exec(params, ProcessTelemetry::default()) .await .expect("start process with proxy"); let output = backend @@ -1379,7 +1438,9 @@ mod tests { let mut params = test_exec_params(HashMap::new()); params.cwd = cwd; - let result = LocalProcess::default().start_process(params).await; + let result = LocalProcess::default() + .start_process(params, ProcessTelemetry::default()) + .await; let Err(error) = result else { panic!("non-native cwd should be rejected"); }; @@ -1406,7 +1467,7 @@ mod tests { params.process_id = ProcessId::from(process_id); params.network_proxy = Some(proxy.clone()); let error = LocalProcess::default() - .start_process(params) + .start_process(params, ProcessTelemetry::default()) .await .err() .expect("invalid callback process ID should be rejected"); @@ -1419,7 +1480,7 @@ mod tests { boundary.network_proxy = Some(proxy); boundary.argv.clear(); let error = LocalProcess::default() - .start_process(boundary) + .start_process(boundary, ProcessTelemetry::default()) .await .err() .expect("valid boundary process ID should proceed to process preparation"); @@ -1440,7 +1501,7 @@ mod tests { ordinary.process_id = ProcessId::from(process_id); ordinary.argv.clear(); let error = LocalProcess::default() - .start_process(ordinary) + .start_process(ordinary, ProcessTelemetry::default()) .await .err() .expect("empty argv should be rejected after ID validation"); @@ -1965,6 +2026,7 @@ mod tests { exit_rx, Arc::clone(&backend.inner), output_notify, + ProcessTelemetry::default(), )); TestProcess { diff --git a/codex-rs/exec-server/src/noise_relay/executor_stream.rs b/codex-rs/exec-server/src/noise_relay/executor_stream.rs index 32c7c0c1d7..612cc9781f 100644 --- a/codex-rs/exec-server/src/noise_relay/executor_stream.rs +++ b/codex-rs/exec-server/src/noise_relay/executor_stream.rs @@ -23,6 +23,7 @@ use crate::noise_relay::take_next_sequence; use crate::relay::encode_relay_message_frame; use crate::relay_proto::RelayData; use crate::relay_proto::RelayMessageFrame; +use crate::telemetry::ExecutorRegistration; /// Identifies one completed virtual-stream instance. /// @@ -91,6 +92,7 @@ pub(crate) fn spawn_noise_virtual_stream( physical_outgoing_tx: mpsc::Sender>, closed_stream_tx: mpsc::Sender, transport: NoiseTransport, + executor_registration: Option>, ) -> NoiseVirtualStream { let (outgoing_tx, mut outgoing_rx) = mpsc::channel(CHANNEL_CAPACITY); let (incoming_tx, incoming_rx) = mpsc::channel(CHANNEL_CAPACITY); @@ -162,6 +164,7 @@ pub(crate) fn spawn_noise_virtual_stream( incoming_rx, disconnected_rx, writer_task, + executor_registration, }; tokio::spawn(async move { handler.run_connection(connection).await; diff --git a/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs b/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs index 986fd650fe..4bf8f945df 100644 --- a/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/executor_stream_tests.rs @@ -99,6 +99,7 @@ async fn processor_exit_reports_closed_virtual_stream() -> Result<()> { physical_outgoing_tx, closed_stream_tx, executor_transport, + /*executor_registration*/ None, ); let message = JSONRPCMessage::Response(JSONRPCResponse { diff --git a/codex-rs/exec-server/src/noise_relay/forward_stream_tests.rs b/codex-rs/exec-server/src/noise_relay/forward_stream_tests.rs index 25deb383ed..d19c3232a5 100644 --- a/codex-rs/exec-server/src/noise_relay/forward_stream_tests.rs +++ b/codex-rs/exec-server/src/noise_relay/forward_stream_tests.rs @@ -59,6 +59,7 @@ async fn forwards_opaque_noise_payloads_and_drains_before_closing() -> Result<() physical_outgoing_tx, closed_stream_tx, executor_transport, + /*executor_registration*/ None, ); let (socket, _) = timeout(TEST_TIMEOUT, destination.accept()).await??; let mut socket = timeout(TEST_TIMEOUT, accept_async(socket)).await??; diff --git a/codex-rs/exec-server/src/noise_relay/stream_handler.rs b/codex-rs/exec-server/src/noise_relay/stream_handler.rs index a85d5ca7f7..b0c610933c 100644 --- a/codex-rs/exec-server/src/noise_relay/stream_handler.rs +++ b/codex-rs/exec-server/src/noise_relay/stream_handler.rs @@ -1,4 +1,5 @@ use std::future::Future; +use std::sync::Arc; use bytes::Bytes; use codex_exec_server_protocol::JSONRPCMessage; @@ -13,13 +14,14 @@ use crate::connection::JsonRpcConnectionEvent; use crate::connection::JsonRpcTransport; use crate::noise_relay::message_framing::frame_jsonrpc_message; use crate::server::ConnectionProcessor; -use crate::telemetry::ConnectionTransport; +use crate::telemetry::ExecutorRegistration; pub(crate) struct NoiseStreamConnection { pub(crate) outgoing_tx: mpsc::Sender, pub(crate) incoming_rx: mpsc::Receiver, pub(crate) disconnected_rx: watch::Receiver, pub(crate) writer_task: JoinHandle<()>, + pub(crate) executor_registration: Option>, } pub(crate) struct NoiseOutboundMessage { @@ -67,7 +69,7 @@ impl NoiseStreamHandler for ConnectionProcessor { self, connection: NoiseStreamConnection, ) { - ConnectionProcessor::run_connection( + ConnectionProcessor::run_registered_connection( &self, JsonRpcConnection { outgoing_tx: connection.outgoing_tx, @@ -76,7 +78,7 @@ impl NoiseStreamHandler for ConnectionProcessor { task_handles: vec![connection.writer_task], transport: JsonRpcTransport::Plain, }, - ConnectionTransport::Relay, + connection.executor_registration, ) .await; } diff --git a/codex-rs/exec-server/src/process_sandbox.rs b/codex-rs/exec-server/src/process_sandbox.rs index 0bf5853a1d..99d96a7339 100644 --- a/codex-rs/exec-server/src/process_sandbox.rs +++ b/codex-rs/exec-server/src/process_sandbox.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use std::sync::Arc; +use crate::process_telemetry::ProcessTelemetry; use codex_exec_server_protocol::JSONRPCErrorError; use codex_network_proxy::CUSTOM_CA_ENV_KEYS; use codex_network_proxy::ManagedNetworkSandboxContext; @@ -75,12 +76,13 @@ impl PreparedExecRequest { } } -pub(crate) async fn prepare_exec_request( +pub(crate) async fn prepare_exec_request_with_telemetry( params: &ExecParams, env: HashMap, runtime_paths: Option<&ExecServerRuntimePaths>, network_policy_decider: Option>, network_policy_audit_observer: Option, + telemetry: &ProcessTelemetry, ) -> Result { #[cfg(target_os = "windows")] let mut env = env; @@ -105,6 +107,7 @@ pub(crate) async fn prepare_exec_request( env, network_policy_decider, network_policy_audit_observer, + telemetry, ) .await?; let Some(sandbox_context) = params.sandbox.as_ref() else { @@ -299,6 +302,7 @@ async fn prepare_managed_network( env: HashMap, network_policy_decider: Option>, network_policy_audit_observer: Option, + telemetry: &ProcessTelemetry, ) -> Result< ( HashMap, @@ -316,6 +320,20 @@ async fn prepare_managed_network( if let Some(observer) = network_policy_audit_observer { state.set_policy_audit_observer(observer); } + if let Some(launch_context) = &telemetry.launch_context { + state.set_launch_span_context(launch_context.clone()); + } + state.set_process_log_metadata(codex_network_proxy::NetworkProxyProcessLogMetadata { + thread_id: telemetry.thread_id.clone(), + tool_call_id: telemetry.tool_call_id.clone(), + executor_identity: telemetry + .executor_registration + .as_ref() + .map(|registration| codex_network_proxy::ExecutorLogIdentity { + environment_id: registration.environment_id.clone(), + registration_id: registration.executor_registration_id.clone(), + }), + }); let mut builder = NetworkProxy::builder().state(Arc::new(state)); if let Some(network_policy_decider) = network_policy_decider { builder = builder.policy_decider_arc(network_policy_decider); diff --git a/codex-rs/exec-server/src/process_sandbox_tests.rs b/codex-rs/exec-server/src/process_sandbox_tests.rs index a66387ee0d..00840569c1 100644 --- a/codex-rs/exec-server/src/process_sandbox_tests.rs +++ b/codex-rs/exec-server/src/process_sandbox_tests.rs @@ -1,9 +1,13 @@ use std::collections::HashMap; use std::net::SocketAddr; +use std::sync::Arc; use std::time::Duration; +use codex_exec_server_protocol::JSONRPCErrorError; #[cfg(target_os = "macos")] use codex_network_proxy::ManagedNetworkSandboxContext; +use codex_network_proxy::NetworkPolicyAuditObserver; +use codex_network_proxy::NetworkPolicyDecider; use codex_network_proxy::NetworkProxyConfig; use codex_network_proxy::PROXY_ATTRIBUTION_TOKEN_ENV_KEY; use codex_network_proxy::RemoteNetworkProxyConfig; @@ -23,15 +27,34 @@ use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; use tokio::time::timeout; -use super::prepare_exec_request; +use super::PreparedExecRequest; +use super::prepare_exec_request_with_telemetry; #[cfg(unix)] use crate::CODEX_ARG0_EXEC_HELPER_ARG1; use crate::ExecParams; -#[cfg(any(unix, windows))] use crate::ExecServerRuntimePaths; #[cfg(any(unix, windows))] use crate::FileSystemSandboxContext; use crate::ProcessId; +use crate::process_telemetry::ProcessTelemetry; + +async fn prepare_exec_request( + params: &ExecParams, + env: HashMap, + runtime_paths: Option<&ExecServerRuntimePaths>, + network_policy_decider: Option>, + network_policy_audit_observer: Option, +) -> Result { + prepare_exec_request_with_telemetry( + params, + env, + runtime_paths, + network_policy_decider, + network_policy_audit_observer, + &ProcessTelemetry::default(), + ) + .await +} #[cfg(unix)] #[tokio::test] @@ -49,6 +72,7 @@ async fn sandbox_request_wraps_native_argv_on_executor() { cwd_uri.clone(), ); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-1"), argv: vec![ "/bin/bash".to_string(), @@ -123,6 +147,7 @@ async fn sandbox_request_routes_custom_arg0_to_inner_helper() { cwd_uri.clone(), ); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-custom-arg0"), argv: vec!["/bin/sh".to_string(), "-c".to_string(), "true".to_string()], cwd: cwd_uri, @@ -185,6 +210,7 @@ async fn sandbox_request_allows_prepared_managed_proxy_port() { cwd_uri.clone(), ); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-managed-network"), argv: vec!["/usr/bin/true".to_string()], cwd: cwd_uri, @@ -230,6 +256,7 @@ async fn native_request_preserves_native_launch_fields() { let cwd_uri = PathUri::from_abs_path(&cwd); let env = HashMap::from([("TEST_ENV".to_string(), "value".to_string())]); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-1"), argv: vec!["echo".to_string(), "hello".to_string()], cwd: cwd_uri, @@ -275,6 +302,7 @@ async fn native_request_handles_remote_proxy_config_for_platform() { let proxy_config = RemoteNetworkProxyConfig::from_effective_config(&config) .expect("supported remote proxy config"); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-remote-proxy"), argv: vec!["echo".to_string(), "hello".to_string()], cwd: PathUri::from_abs_path(&cwd), @@ -358,6 +386,7 @@ async fn disabled_remote_proxy_config_is_rejected_before_exporting_ports() { RemoteNetworkProxyConfig::from_effective_config(&NetworkProxyConfig::default()) .expect("serializable disabled proxy config"); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-disabled-remote-proxy"), argv: vec!["echo".to_string(), "hello".to_string()], cwd: PathUri::from_abs_path(&cwd), @@ -419,6 +448,7 @@ async fn managed_network_honors_windows_sandbox_level(windows_sandbox_level: Win }) .expect("supported remote proxy config"); let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("process-managed-network"), argv: vec!["cmd.exe".to_string(), "/c".to_string(), "exit".to_string()], cwd: cwd_uri, diff --git a/codex-rs/exec-server/src/process_telemetry.rs b/codex-rs/exec-server/src/process_telemetry.rs new file mode 100644 index 0000000000..f926fa9889 --- /dev/null +++ b/codex-rs/exec-server/src/process_telemetry.rs @@ -0,0 +1,78 @@ +//! Emits bounded process lifecycle telemetry using identity captured at launch. +//! Reconnects and later process operations must not replace that identity. + +use std::sync::Arc; + +use codex_sandboxing::SandboxType; +use opentelemetry::trace::SpanContext; + +use crate::telemetry::ExecutorRegistration; + +/// Log fields captured at launch, never refreshed from a resumed session. +#[derive(Clone, Default)] +pub(crate) struct ProcessTelemetry { + pub(crate) launch_context: Option, + pub(crate) thread_id: Option, + pub(crate) tool_call_id: Option, + pub(crate) executor_registration: Option>, +} + +/// Lifecycle events with outcome fields only when a process has exited. +pub(crate) enum ProcessTelemetryEvent { + Start, + SpawnFailed, + SandboxDenied, + Exit { + exit_code: i32, + termination_requested: bool, + }, +} + +impl ProcessTelemetry { + pub(crate) fn log(&self, event: ProcessTelemetryEvent, sandbox: SandboxType) { + let (event_name, exit_code, termination_requested, reason) = match event { + ProcessTelemetryEvent::Start => ("codex.exec_server.process_start", None, None, None), + ProcessTelemetryEvent::SpawnFailed => { + ("codex.exec_server.process_spawn_failed", None, None, None) + } + ProcessTelemetryEvent::SandboxDenied => ( + "codex.exec_server.sandbox_denied", + None, + None, + Some("inferred_denial"), + ), + ProcessTelemetryEvent::Exit { + exit_code, + termination_requested, + } => ( + "codex.exec_server.process_exit", + Some(exit_code), + Some(termination_requested), + None, + ), + }; + let trace_id = self + .launch_context + .as_ref() + .map(|span| span.trace_id().to_string()); + let span_id = self + .launch_context + .as_ref() + .map(|span| span.span_id().to_string()); + tracing::event!( + target: "codex_otel.log_only", + tracing::Level::INFO, + event.name = event_name, + launch.trace_id = trace_id.as_deref(), + launch.span_id = span_id.as_deref(), + conversation.id = self.thread_id.as_deref(), + tool.call_id = self.tool_call_id.as_deref(), + executor.environment_id = self.executor_registration.as_ref().map(|registration| registration.environment_id.as_str()), + executor.registration_id = self.executor_registration.as_ref().map(|registration| registration.executor_registration_id.as_str()), + sandbox.type = ?sandbox, + process.exit_code = exit_code, + process.termination_requested = termination_requested, + reason, + ); + } +} diff --git a/codex-rs/exec-server/src/relay.rs b/codex-rs/exec-server/src/relay.rs index f53a09a3e2..40afbf0e28 100644 --- a/codex-rs/exec-server/src/relay.rs +++ b/codex-rs/exec-server/src/relay.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::sync::Arc; use std::time::Duration; use codex_exec_server_protocol::JSONRPCMessage; @@ -41,6 +42,7 @@ use crate::relay_proto::RelayResume; use crate::relay_proto::relay_message_frame; #[cfg(test)] use crate::server::ConnectionProcessor; +use crate::telemetry::ExecutorRegistration; use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT; use crate::websocket_pong_watchdog::WEBSOCKET_PONG_TIMEOUT_REASON; use crate::websocket_pong_watchdog::WebSocketPongWatchdog; @@ -495,6 +497,9 @@ where environment_id, executor_registration_id, "Noise executor relay details" ); + let executor_registration = + ExecutorRegistration::new(environment_id.clone(), executor_registration_id.clone()) + .map(Arc::new); let (mut websocket_sink, mut websocket_stream) = stream.split(); let (physical_outgoing_tx, mut physical_outgoing_rx) = mpsc::channel::>(CHANNEL_CAPACITY); @@ -686,6 +691,7 @@ where physical_outgoing_tx.clone(), closed_stream_tx.clone(), transport, + executor_registration.clone(), ), ); } diff --git a/codex-rs/exec-server/src/relay_noise_tests.rs b/codex-rs/exec-server/src/relay_noise_tests.rs index 99eda6de8a..ff402b316d 100644 --- a/codex-rs/exec-server/src/relay_noise_tests.rs +++ b/codex-rs/exec-server/src/relay_noise_tests.rs @@ -32,6 +32,9 @@ use crate::noise_channel::NoiseChannelPublicKey; use crate::noise_channel::noise_channel_prologue; use crate::noise_relay::NoiseHarnessConnectionArgs; use crate::noise_relay::noise_harness_connection_from_websocket; +use crate::noise_relay::stream_handler::NoiseOutboundMessage; +use crate::noise_relay::stream_handler::NoiseStreamConnection; +use crate::noise_relay::stream_handler::NoiseStreamHandler; use crate::relay::RelayFrameBodyKind; use crate::relay::decode_relay_message_frame; use crate::relay::encode_relay_message_frame; @@ -41,6 +44,40 @@ use crate::server::ConnectionProcessor; const ENVIRONMENT_ID: &str = "environment-1"; const EXECUTOR_REGISTRATION_ID: &str = "registration-1"; +#[derive(Clone)] +struct ObservedRegistration( + ConnectionProcessor, + tokio::sync::mpsc::Sender>, +); + +impl NoiseStreamHandler for ObservedRegistration { + type Incoming = JsonRpcConnectionEvent; + type Outgoing = JSONRPCMessage; + + fn decode(payload: bytes::Bytes) -> Result { + ConnectionProcessor::decode(payload) + } + fn encode(message: Self::Outgoing) -> Result { + ConnectionProcessor::encode(message) + } + async fn run_connection( + self, + connection: NoiseStreamConnection, + ) { + let registration = connection + .executor_registration + .as_ref() + .map(|registration| { + ( + registration.environment_id.clone(), + registration.executor_registration_id.clone(), + ) + }); + let _ = self.1.send(registration).await; + NoiseStreamHandler::run_connection(self.0, connection).await; + } +} + #[tokio::test] async fn missing_pong_disconnects_physical_relay() -> Result<()> { let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -145,6 +182,7 @@ impl HarnessKeyValidator for BlockingValidator { #[tokio::test] async fn processor_exit_resets_noise_harness_stream() -> Result<()> { + let (registration_tx, mut registration_rx) = tokio::sync::mpsc::channel(1); let listener = TcpListener::bind("127.0.0.1:0").await?; let connecting = tokio::spawn(connect_async(format!("ws://{}", listener.local_addr()?))); let (socket, _) = listener.accept().await?; @@ -155,10 +193,13 @@ async fn processor_exit_resets_noise_harness_stream() -> Result<()> { release.notify_one(); let environment_task = AbortOnDropHandle::new(tokio::spawn(run_multiplexed_environment( environment_websocket, - ConnectionProcessor::new(ExecServerRuntimePaths::new( - std::env::current_exe()?, - /*codex_linux_sandbox_exe*/ None, - )?), + ObservedRegistration( + ConnectionProcessor::new(ExecServerRuntimePaths::new( + std::env::current_exe()?, + /*codex_linux_sandbox_exe*/ None, + )?), + registration_tx, + ), ENVIRONMENT_ID.to_string(), EXECUTOR_REGISTRATION_ID.to_string(), identity.clone(), @@ -187,6 +228,13 @@ async fn processor_exit_resets_noise_harness_stream() -> Result<()> { result: serde_json::Value::Null, })) .await?; + assert_eq!( + timeout(Duration::from_secs(1), registration_rx.recv()).await?, + Some(Some(( + ENVIRONMENT_ID.to_string(), + EXECUTOR_REGISTRATION_ID.to_string() + ))), + ); assert!(matches!( timeout(Duration::from_secs(1), connection.incoming_rx.recv()).await?, Some(JsonRpcConnectionEvent::Disconnected { reason: Some(reason) }) diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 83d4378344..b6eb30adba 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -16,6 +16,7 @@ use codex_exec_server_protocol::JSONRPCNotification; use codex_exec_server_protocol::JSONRPCRequest; use codex_exec_server_protocol::JSONRPCResponse; use codex_exec_server_protocol::RequestId; +use codex_protocol::protocol::W3cTraceContext; use serde::Serialize; use serde::de::DeserializeOwned; use serde_json::Value; @@ -208,14 +209,26 @@ where R: Serialize + Send + 'static, F: Fn(Arc, P) -> Fut + Send + Sync + 'static, Fut: Future> + Send + 'static, + { + self.request_with_trace(method, move |state, params, _trace| handler(state, params)); + } + + /// Supplies the incoming W3C carrier to handlers that need it without requiring a trace exporter. + pub(crate) fn request_with_trace(&mut self, method: &'static str, handler: F) + where + P: DeserializeOwned + Send + 'static, + R: Serialize + Send + 'static, + F: Fn(Arc, P, Option) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, { self.request_routes.insert( method, Box::new(move |state, request| { + let trace = request.trace; let request_id = request.id; let params = request.params; let response = - decode_request_params::

(params).map(|params| handler(state, params)); + decode_request_params::

(params).map(|params| handler(state, params, trace)); Box::pin(async move { let response = match response { Ok(response) => response.await, diff --git a/codex-rs/exec-server/src/server.rs b/codex-rs/exec-server/src/server.rs index 15a5c877cf..aea6d6a73f 100644 --- a/codex-rs/exec-server/src/server.rs +++ b/codex-rs/exec-server/src/server.rs @@ -8,6 +8,10 @@ mod request_dispatcher; mod session_registry; mod transport; +#[cfg(all(test, unix))] +#[path = "server/process_otel_tests.rs"] +mod process_otel_tests; + pub(crate) use handler::ExecServerHandler; pub(crate) use processor::ConnectionProcessor; pub use request_dispatcher::ConcurrentRequestLimit; diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 92fcd547be..feb0f5b93a 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -6,6 +6,7 @@ use std::sync::atomic::Ordering; use codex_exec_server_protocol::JSONRPCErrorError; use codex_exec_server_protocol::RequestId; use codex_http_client::HttpClientFactory; +use opentelemetry::trace::SpanContext; use serde_json::to_value; use std::collections::HashSet; use tokio::sync::Mutex; @@ -69,8 +70,10 @@ use crate::rpc::invalid_request; use crate::server::file_system_handler::FileSystemHandler; use crate::server::session_registry::SessionHandle; use crate::server::session_registry::SessionRegistry; +use crate::telemetry::ExecutorRegistration; pub(crate) struct ExecServerHandler { + pub(super) executor_registration: Option>, session_registry: Arc, notifications: RpcNotificationSender, session: StdMutex>, @@ -92,6 +95,7 @@ impl ExecServerHandler { http_client_factory: HttpClientFactory, ) -> Self { Self { + executor_registration: None, session_registry, notifications, session: StdMutex::new(None), @@ -172,9 +176,23 @@ impl ExecServerHandler { Ok(()) } - pub(crate) async fn exec(&self, params: ExecParams) -> Result { + pub(crate) async fn exec( + &self, + params: ExecParams, + launch_context: Option, + ) -> Result { let session = self.require_initialized_for("exec")?; - session.process().exec(params).await + session + .process() + .exec( + params, + crate::process_telemetry::ProcessTelemetry { + launch_context, + executor_registration: self.executor_registration.clone(), + ..Default::default() + }, + ) + .await } pub(crate) fn environment_info(&self) -> Result { diff --git a/codex-rs/exec-server/src/server/handler/tests.rs b/codex-rs/exec-server/src/server/handler/tests.rs index 1fb96850ed..5d2f43d11a 100644 --- a/codex-rs/exec-server/src/server/handler/tests.rs +++ b/codex-rs/exec-server/src/server/handler/tests.rs @@ -27,6 +27,7 @@ fn exec_params(process_id: &str) -> ExecParams { fn exec_params_with_argv(process_id: &str, argv: Vec) -> ExecParams { ExecParams { + metadata: Default::default(), process_id: ProcessId::from(process_id), argv, cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) @@ -116,8 +117,8 @@ async fn duplicate_process_ids_allow_only_one_successful_start() { let second_handler = Arc::clone(&handler); let (first, second) = tokio::join!( - first_handler.exec(exec_params("proc-1")), - second_handler.exec(exec_params("proc-1")), + first_handler.exec(exec_params("proc-1"), /*launch_context*/ None), + second_handler.exec(exec_params("proc-1"), /*launch_context*/ None), ); let (successes, failures): (Vec<_>, Vec<_>) = @@ -141,7 +142,7 @@ async fn duplicate_process_ids_allow_only_one_successful_start() { async fn terminate_reports_false_after_process_exit() { let handler = initialized_handler().await; handler - .exec(exec_params("proc-1")) + .exec(exec_params("proc-1"), /*launch_context*/ None) .await .expect("start process"); @@ -188,10 +189,13 @@ async fn long_poll_read_fails_after_session_resume() { // Keep the process quiet and alive so the pending read can only complete // after session resume, not because the process produced output or exited. first_handler - .exec(exec_params_with_argv( - "proc-long-poll", - shell_argv("sleep 5", "ping -n 6 127.0.0.1 >NUL"), - )) + .exec( + exec_params_with_argv( + "proc-long-poll", + shell_argv("sleep 5", "ping -n 6 127.0.0.1 >NUL"), + ), + /*launch_context*/ None, + ) .await .expect("start process"); @@ -306,13 +310,16 @@ async fn output_and_exit_are_retained_after_notification_receiver_closes() { let process_id = ProcessId::from("proc-notification-fail"); handler - .exec(exec_params_with_argv( - process_id.as_str(), - shell_argv( - "sleep 0.05; printf 'first\\n'; sleep 0.05; printf 'second\\n'", - "echo first&& ping -n 2 127.0.0.1 >NUL&& echo second", + .exec( + exec_params_with_argv( + process_id.as_str(), + shell_argv( + "sleep 0.05; printf 'first\\n'; sleep 0.05; printf 'second\\n'", + "echo first&& ping -n 2 127.0.0.1 >NUL&& echo second", + ), ), - )) + /*launch_context*/ None, + ) .await .expect("start process"); @@ -324,7 +331,10 @@ async fn output_and_exit_are_retained_after_notification_receiver_closes() { tokio::time::sleep(Duration::from_millis(100)).await; handler - .exec(exec_params(process_id.as_str())) + .exec( + exec_params(process_id.as_str()), + /*launch_context*/ None, + ) .await .expect("process id should be reusable after exit retention"); diff --git a/codex-rs/exec-server/src/server/process_handler.rs b/codex-rs/exec-server/src/server/process_handler.rs index 4d933afbe1..a4aca9d72c 100644 --- a/codex-rs/exec-server/src/server/process_handler.rs +++ b/codex-rs/exec-server/src/server/process_handler.rs @@ -1,3 +1,4 @@ +use crate::process_telemetry::ProcessTelemetry; use codex_exec_server_protocol::JSONRPCErrorError; use crate::ExecServerRuntimePaths; @@ -39,8 +40,12 @@ impl ProcessHandler { self.process.set_notification_sender(notifications); } - pub(crate) async fn exec(&self, params: ExecParams) -> Result { - self.process.exec(params).await + pub(crate) async fn exec( + &self, + params: ExecParams, + telemetry: ProcessTelemetry, + ) -> Result { + self.process.exec(params, telemetry).await } pub(crate) async fn exec_read( diff --git a/codex-rs/exec-server/src/server/process_otel_tests.rs b/codex-rs/exec-server/src/server/process_otel_tests.rs new file mode 100644 index 0000000000..3460408ea5 --- /dev/null +++ b/codex-rs/exec-server/src/server/process_otel_tests.rs @@ -0,0 +1,752 @@ +//! Verifies exported lifecycle and network logs retain launch identity without private payloads. + +#![allow(clippy::expect_used)] + +use std::collections::BTreeMap; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use codex_exec_server_protocol::JSONRPCMessage; +use codex_exec_server_protocol::JSONRPCRequest; +use codex_exec_server_protocol::RequestId; +use codex_http_client::HttpClientFactory; +use codex_http_client::OutboundProxyPolicy; +use codex_network_proxy::NetworkProxyConfig; +use codex_network_proxy::RemoteNetworkProxyConfig; +use codex_network_proxy::RemoteNetworkProxyLaunchConfig; +use codex_otel::OtelExporter; +use codex_otel::OtelHttpProtocol; +use codex_otel::OtelProvider; +use codex_otel::OtelSettings; +use codex_protocol::protocol::W3cTraceContext; +use codex_utils_path_uri::PathUri; +use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncWriteExt; +use tokio::sync::mpsc; +use tracing::Instrument; +use tracing_subscriber::prelude::*; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; + +use super::ExecServerHandler; +use super::registry::build_router; +use super::session_registry::SessionRegistry; +use crate::ExecServerRuntimePaths; +use crate::ExecServerTelemetry; +use crate::connection::JsonRpcConnectionEvent; +use crate::protocol::EXEC_METHOD; +use crate::protocol::EXEC_READ_METHOD; +use crate::protocol::EXEC_TERMINATE_METHOD; +use crate::protocol::InitializeParams; +use crate::protocol::ReadResponse; +use crate::rpc::RpcNotificationSender; +use crate::rpc::RpcRouter; +use crate::rpc::RpcServerOutboundMessage; +use crate::telemetry::ExecutorRegistration; + +const PRIVATE_PAYLOAD: &str = "private-process-payload"; +const TRACE_ID: &str = "11111111111111111111111111111111"; +const LAUNCH_SPANS: [&str; 2] = ["2222222222222222", "3333333333333333"]; +const THREADS: [&str; 2] = [ + "11111111-1111-4111-8111-111111111111", + "22222222-2222-4222-8222-222222222222", +]; +const CALLS: [&str; 2] = ["call-first", "call-second"]; +const LATER_TRACE: &str = "00-44444444444444444444444444444444-5555555555555555-01"; + +#[test] +fn exported_process_and_network_logs_keep_the_validated_launch_reference() { + // Keep the subscriber on the runtime's only thread, including proxy background tasks. + for (export_traces, registered) in [(false, true), (true, true), (false, false)] { + let records = exported_logs(export_traces, |runtime, _, _| { + runtime.block_on(async { + let server = TestServer::new(); + let mut handlers = Vec::new(); + let mut session_ids = Vec::new(); + for client_name in ["first-orchestrator", "second-orchestrator"] { + let (handler, session_id) = server.new_orchestrator( + client_name, registered.then_some("original"), + ).await; + handlers.push(handler); + session_ids.push(session_id); + } + let proxy = RemoteNetworkProxyLaunchConfig::new( + RemoteNetworkProxyConfig::from_effective_config(&NetworkProxyConfig { + enabled: true, + ..NetworkProxyConfig::default() + }).expect("proxy config"), + ); + // Unsampled incoming context must still correlate logs-only export. + let flags = if export_traces { "01" } else { "00" }; + let launches = LAUNCH_SPANS.map(|span| format!("00-{TRACE_ID}-{span}-{flags}")); + for (batch, traces) in [ + [Some(launches[0].as_str()), Some(launches[1].as_str())], + [None, Some(PRIVATE_PAYLOAD)], + ].into_iter().enumerate() { + // Process IDs are session-scoped and deliberately reused across clients. + let process_id = format!("{PRIVATE_PAYLOAD}-{batch}"); + for (index, (handler, trace)) in handlers.iter().zip(traces).enumerate() { + let mut params = process_start_params( + &process_id, + json!(["/bin/sh", "-c", "printf '%s\\n' \"$HTTP_PROXY\"; read ignored", PRIVATE_PAYLOAD]), + PathUri::from_host_native_path(std::env::current_dir().expect("cwd")).expect("cwd URI"), + ); + params["metadata"] = if batch == 0 { + process_metadata(index) + } else { + json!({"toolCallId": if index == 0 { None } else { Some("private-process-payload\n") }}) + }; + params["metadata"]["executorRegistrationId"] = json!(PRIVATE_PAYLOAD); + params["metadata"]["environmentId"] = json!(PRIVATE_PAYLOAD); + params["executorRegistrationId"] = json!(PRIVATE_PAYLOAD); + params["environmentId"] = json!(PRIVATE_PAYLOAD); + params["pipeStdin"] = json!(true); + params["networkProxy"] = json!(proxy); + request(&server.router, handler, EXEC_METHOD, params, trace).await; + } + if batch == 0 { + handlers[0].shutdown().await; + let resumed = new_handler( + &server.sessions, server.outgoing.clone(), registered.then_some("resumed"), + ); + resumed.initialize(InitializeParams { + client_name: "resumed-orchestrator".to_string(), + resume_session_id: Some(session_ids[0].clone()), + }).await.expect("resume session"); + resumed.initialized().expect("initialized resumed orchestrator"); + handlers[0] = resumed; + } + for handler in &handlers { + let output: ReadResponse = serde_json::from_value(request( + &server.router, handler, EXEC_READ_METHOD, + json!({"processId": process_id, "waitMs": 1000}), Some(LATER_TRACE), + ).await).expect("read proxy address"); + let proxy_address = String::from_utf8(output.chunks.into_iter() + .flat_map(|chunk| chunk.chunk.into_inner()).collect()).expect("UTF-8 output"); + let mut connection = tokio::net::TcpStream::connect( + proxy_address.trim().strip_prefix("http://").expect("HTTP proxy address"), + ).await.expect("connect to process proxy"); + connection.write_all(b"CONNECT 8.8.8.8:443 HTTP/1.1\r\nHost: 8.8.8.8:443\r\n\r\n") + .await.expect("request denied destination"); + let mut response = [0_u8; 256]; + let length = tokio::time::timeout(Duration::from_secs(5), connection.read(&mut response)) + .await.expect("proxy response timeout").expect("proxy response"); + assert!(String::from_utf8_lossy(&response[..length]).starts_with("HTTP/1.1 403")); + request(&server.router, handler, EXEC_TERMINATE_METHOD, + json!({"processId": process_id}), Some(LATER_TRACE)).await; + read_until_closed(&server.router, handler, &process_id).await; + } + } + server.sessions.shutdown().await; + }); + }); + let mut events = Vec::new(); + for record in &records { + let name = attribute(record, "event.name").expect("event name"); + let trace = attribute(record, "launch.trace_id"); + let span = attribute(record, "launch.span_id"); + if export_traces + && name.starts_with("codex.exec_server.process_") + && let Some(span) = span + { + assert_eq!(record["traceId"].as_str(), trace); + assert_ne!(record["spanId"].as_str().expect("native child span"), span); + } + events.push(( + name.to_string(), + trace.map(str::to_string), + span.map(str::to_string), + attribute(record, "conversation.id").map(str::to_string), + attribute(record, "tool.call_id").map(str::to_string), + attribute(record, "executor.environment_id").map(str::to_string), + attribute(record, "executor.registration_id").map(str::to_string), + )); + } + let mut expected = Vec::new(); + for (index, span) in [Some(LAUNCH_SPANS[0]), Some(LAUNCH_SPANS[1]), None, None] + .into_iter() + .enumerate() + { + for name in [ + "codex.exec_server.process_start", + "codex.exec_server.process_exit", + "codex.network_proxy.policy_decision", + ] { + expected.push(( + name.to_string(), + span.map(|_| TRACE_ID.to_string()), + span.map(str::to_string), + (index < 2).then(|| THREADS[index].to_string()), + (index < 2).then(|| CALLS[index].to_string()), + registered.then(|| "environment".to_string()), + registered.then(|| if index == 2 { "resumed" } else { "original" }.to_string()), + )); + } + } + events.sort(); + expected.sort(); + assert_eq!(events, expected); + } +} + +#[test] +fn launch_rpc_span_finishes_before_its_process_exits() { + let mut observed = None; + let records = exported_logs(/*export_traces*/ true, |runtime, otel, collector| { + runtime.block_on(async { + let server = TestServer::new(); + let (handler, _) = server + .new_orchestrator("process-span-orchestrator", Some("original")) + .await; + let launch = format!("00-{TRACE_ID}-{}-01", LAUNCH_SPANS[0]); + let mut params = process_start_params( + PRIVATE_PAYLOAD, + json!(["/bin/sh", "-c", "read ignored", PRIVATE_PAYLOAD]), + PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) + .expect("cwd URI"), + ); + params["metadata"] = process_metadata(/*index*/ 0); + params["pipeStdin"] = json!(true); + request(&server.router, &handler, EXEC_METHOD, params, Some(&launch)).await; + + // Flush completed spans while the process is still waiting on its open stdin. + // The old task's in_current_span() retains the RPC span, so this snapshot lacks it. + let spans_before_exit = flushed_spans(otel, collector).await; + let running: ReadResponse = serde_json::from_value( + request( + &server.router, + &handler, + EXEC_READ_METHOD, + json!({"processId": PRIVATE_PAYLOAD, "waitMs": 0}), + Some(LATER_TRACE), + ) + .await, + ) + .expect("read running process"); + request( + &server.router, + &handler, + EXEC_TERMINATE_METHOD, + json!({"processId": PRIVATE_PAYLOAD}), + Some(LATER_TRACE), + ) + .await; + let exited = read_until_closed(&server.router, &handler, PRIVATE_PAYLOAD).await; + server.sessions.shutdown().await; + let spans_after_exit = flushed_spans(otel, collector).await; + // Finish cleanup before asserting the regression so a failure leaves no child behind. + observed = Some((running, exited, spans_before_exit, spans_after_exit)); + }); + }); + let (running, exited, spans_before_exit, spans_after_exit) = + observed.expect("process observations"); + assert_eq!( + (running.exited, running.closed, running.exit_code), + (false, false, None) + ); + let launch_spans: Vec<_> = spans_before_exit + .iter() + .filter(|span| span["name"] == EXEC_METHOD) + .collect(); + assert_eq!( + launch_spans.len(), + 1, + "process/start span must export before child exit" + ); + let launch_span = launch_spans[0]; + assert_eq!( + ( + launch_span["traceId"].as_str(), + launch_span["parentSpanId"].as_str() + ), + (Some(TRACE_ID), Some(LAUNCH_SPANS[0])) + ); + let exit_record = records + .iter() + .find(|record| attribute(record, "event.name") == Some("codex.exec_server.process_exit")) + .expect("process exit log"); + let exit_span_id = exit_record["spanId"] + .as_str() + .expect("process exit span ID"); + // ProcessMetricGuard has the same span name; select the span that emitted the exit log. + let process_spans: Vec<_> = spans_after_exit + .iter() + .filter(|span| span["spanId"].as_str() == Some(exit_span_id)) + .collect(); + assert_eq!( + process_spans.len(), + 1, + "process completion has its own span" + ); + let process_span = process_spans[0]; + assert_eq!(process_span["name"], "codex.exec_server.process"); + assert_eq!( + ( + process_span["traceId"].as_str(), + process_span["parentSpanId"].as_str() + ), + (Some(TRACE_ID), Some(LAUNCH_SPANS[0])) + ); + assert_ne!(process_span["spanId"], launch_span["spanId"]); + let mut expected_exit = + expected_process_attributes("codex.exec_server.process_exit", /*index*/ 0, "None"); + expected_exit["process.exit_code"] = + json!({"intValue": exited.exit_code.expect("exit code").to_string()}); + expected_exit["process.termination_requested"] = json!({"boolValue": true}); + assert_exported_attributes( + &records, + vec![ + expected_process_attributes( + "codex.exec_server.process_start", + /*index*/ 0, + "None", + ), + expected_exit, + ], + ); + for record in &records { + assert_eq!(record["traceId"].as_str(), Some(TRACE_ID)); + let span = if attribute(record, "event.name") == Some("codex.exec_server.process_start") { + launch_span + } else { + process_span + }; + assert_eq!(record["spanId"], span["spanId"]); + } +} + +async fn flushed_spans(otel: &OtelProvider, collector: &MockServer) -> Vec { + let provider = otel + .tracer_provider + .as_ref() + .expect("trace exporter") + .clone(); + // The collector uses this runtime, so the synchronous export flush must not block it. + tokio::time::timeout( + Duration::from_secs(5), + tokio::task::spawn_blocking(move || provider.force_flush()), + ) + .await + .expect("trace flush timeout") + .expect("trace flush task") + .expect("trace flush"); + let mut spans = Vec::new(); + for request in collector + .received_requests() + .await + .expect("exported requests") + { + let body: Value = serde_json::from_slice(&request.body).expect("OTLP JSON"); + let Some(resources) = body["resourceSpans"].as_array() else { + continue; + }; + for resource in resources { + for scope in resource["scopeSpans"].as_array().expect("scope spans") { + spans.extend(scope["spans"].as_array().expect("spans").iter().cloned()); + } + } + } + spans +} + +#[test] +fn exported_spawn_failure_keeps_launch_identity_without_outcome_or_error_text() { + let records = exported_logs(/*export_traces*/ false, |runtime, _, _| { + runtime.block_on(async { + let server = TestServer::new(); + let (handler, _) = server + .new_orchestrator("spawn-failure-orchestrator", Some("original")) + .await; + let directory = tempfile::tempdir().expect("process directory"); + let missing_executable = directory.path().join(PRIVATE_PAYLOAD); + let launch = format!("00-{TRACE_ID}-{}-00", LAUNCH_SPANS[0]); + let mut params = process_start_params( + PRIVATE_PAYLOAD, + json!([missing_executable, PRIVATE_PAYLOAD]), + PathUri::from_host_native_path(directory.path()).expect("cwd URI"), + ); + params["metadata"] = process_metadata(/*index*/ 0); + let response = + request_result(&server.router, &handler, EXEC_METHOD, params, Some(&launch)).await; + let Some(RpcServerOutboundMessage::Error { error, .. }) = response else { + panic!("missing executable must fail to spawn: {response:?}"); + }; + assert_eq!(error.code, -32603); + assert!( + !error.message.is_empty(), + "caller still receives the spawn error" + ); + server.sessions.shutdown().await; + }); + }); + assert_exported_attributes( + &records, + vec![expected_process_attributes( + "codex.exec_server.process_spawn_failed", + /*index*/ 0, + "None", + )], + ); +} + +#[cfg(target_os = "macos")] +#[test] +fn exported_sandbox_denial_keeps_launch_identity_and_separate_exit_outcome() { + let records = exported_logs(/*export_traces*/ false, |runtime, _, _| { + runtime.block_on(async { + let server = TestServer::new(); + let (handler, _) = server.new_orchestrator("sandbox-denial-orchestrator", Some("original")).await; + let directory = tempfile::tempdir().expect("process directory"); + let private_file = directory.path().join(PRIVATE_PAYLOAD); + std::fs::write(&private_file, PRIVATE_PAYLOAD).expect("write test file"); + let cwd = PathUri::from_host_native_path(directory.path()).expect("cwd URI"); + let sandbox = crate::FileSystemSandboxContext::from_legacy_sandbox_policy( + codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), cwd.clone(), + ).expect("read-only sandbox"); + for (index, span) in LAUNCH_SPANS.into_iter().enumerate() { + let process_id = format!("{PRIVATE_PAYLOAD}-{index}"); + let launch = format!("00-{TRACE_ID}-{span}-00"); + let argv = if index == 0 { + json!(["/bin/cat", private_file]) + } else { + // Emit private output, then attempt a real denied write. Exit 23 is produced + // by the child shell only after that write fails, not by sandbox startup. + json!(["/bin/sh", "-c", + "printf '%s' \"$PRIVATE_TEST_VALUE\"; if printf changed > \"$1\"; then exit 0; else exit 23; fi", + PRIVATE_PAYLOAD, private_file]) + }; + let mut params = process_start_params(&process_id, argv, cwd.clone()); + params["metadata"] = process_metadata(index); + params["env"]["LC_ALL"] = json!("C"); + params["sandbox"] = json!(sandbox); + let started: crate::protocol::ExecResponse = serde_json::from_value(request( + &server.router, &handler, EXEC_METHOD, params, Some(&launch), + ).await).expect("start sandboxed process"); + assert_eq!(started.sandbox_type, Some(crate::protocol::ProcessSandboxType::MacosSeatbelt)); + let output = read_until_closed(&server.router, &handler, &process_id).await; + let stdout: Vec = output.chunks.iter() + .filter(|chunk| chunk.stream == crate::protocol::ExecOutputStream::Stdout) + .flat_map(|chunk| chunk.chunk.0.iter().copied()).collect(); + assert_eq!(stdout, PRIVATE_PAYLOAD.as_bytes()); + assert_eq!((output.exit_code, output.sandbox_denied), if index == 0 { + (Some(0), false) + } else { + (Some(23), true) + }); + if index == 1 { + let stderr: Vec = output.chunks.iter() + .filter(|chunk| chunk.stream == crate::protocol::ExecOutputStream::Stderr) + .flat_map(|chunk| chunk.chunk.0.iter().copied()).collect(); + assert!(String::from_utf8_lossy(&stderr).contains("Operation not permitted")); + } + } + assert_eq!(std::fs::read(&private_file).expect("read test file"), PRIVATE_PAYLOAD.as_bytes()); + server.sessions.shutdown().await; + }); + }); + let mut expected = Vec::new(); + for index in 0..2 { + expected.push(expected_process_attributes( + "codex.exec_server.process_start", + index, + "MacosSeatbelt", + )); + let mut exited = + expected_process_attributes("codex.exec_server.process_exit", index, "MacosSeatbelt"); + exited["process.exit_code"] = json!({"intValue": if index == 0 { "0" } else { "23" }}); + exited["process.termination_requested"] = json!({"boolValue": false}); + expected.push(exited); + } + let mut denied = expected_process_attributes( + "codex.exec_server.sandbox_denied", + /*index*/ 1, + "MacosSeatbelt", + ); + denied["reason"] = json!({"stringValue": "inferred_denial"}); + expected.push(denied); + assert_exported_attributes(&records, expected); +} + +async fn read_until_closed( + router: &RpcRouter, + handler: &Arc, + process_id: &str, +) -> ReadResponse { + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let output: ReadResponse = serde_json::from_value( + request( + router, + handler, + EXEC_READ_METHOD, + json!({"processId": process_id, "waitMs": 100}), + Some(LATER_TRACE), + ) + .await, + ) + .expect("read process exit"); + if output.closed { + return output; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("process exits") +} + +fn expected_process_attributes(name: &str, index: usize, sandbox: &str) -> Value { + json!({ + "event.name": {"stringValue": name}, + "launch.trace_id": {"stringValue": TRACE_ID}, + "launch.span_id": {"stringValue": LAUNCH_SPANS[index]}, + "conversation.id": {"stringValue": THREADS[index]}, + "tool.call_id": {"stringValue": CALLS[index]}, + "executor.environment_id": {"stringValue": "environment"}, + "executor.registration_id": {"stringValue": "original"}, + "sandbox.type": {"stringValue": sandbox}, + }) +} + +fn assert_exported_attributes(records: &[Value], mut expected: Vec) { + let mut actual: Vec = records + .iter() + .map(|record| { + Value::Object( + record["attributes"] + .as_array() + .expect("log attributes") + .iter() + .map(|attribute| { + ( + attribute["key"] + .as_str() + .expect("attribute key") + .to_string(), + attribute["value"].clone(), + ) + }) + .collect(), + ) + }) + .collect(); + actual.sort_by_key(Value::to_string); + expected.sort_by_key(Value::to_string); + assert_eq!(actual, expected); +} + +async fn request( + router: &RpcRouter, + handler: &Arc, + method: &str, + params: Value, + traceparent: Option<&str>, +) -> Value { + let response = request_result(router, handler, method, params, traceparent).await; + let Some(RpcServerOutboundMessage::Response { result, .. }) = response else { + panic!("request failed: {response:?}"); + }; + result +} + +async fn request_result( + router: &RpcRouter, + handler: &Arc, + method: &str, + params: Value, + traceparent: Option<&str>, +) -> Option { + let request = JSONRPCRequest { + id: RequestId::Integer(1), + method: method.to_string(), + params: Some(params), + trace: traceparent.map(|traceparent| W3cTraceContext { + traceparent: Some(traceparent.to_string()), + tracestate: None, + }), + }; + let JsonRpcConnectionEvent::QueuedRequest { + request, + request_span, + .. + } = JsonRpcConnectionEvent::message(JSONRPCMessage::Request(request)) + else { + panic!("request span") + }; + let (_, route) = router + .request_route(method) + .expect("registered request route"); + request_span.record("otel.name", method); + route(Arc::clone(handler), request) + .instrument(request_span) + .await +} + +fn attribute<'a>(record: &'a Value, key: &str) -> Option<&'a str> { + record["attributes"] + .as_array() + .expect("log attributes") + .iter() + .find(|attribute| attribute["key"] == key) + .map(|attribute| { + attribute["value"]["stringValue"] + .as_str() + .expect("string attribute") + }) +} + +fn exported_logs( + export_traces: bool, + run: impl FnOnce(&tokio::runtime::Runtime, &OtelProvider, &MockServer), +) -> Vec { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("test runtime"); + let collector = runtime.block_on(MockServer::start()); + runtime.block_on( + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .mount(&collector), + ); + let exporter = OtelExporter::OtlpHttp { + endpoint: format!("{}/v1/logs", collector.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + }; + let otel = OtelProvider::try_new(&OtelSettings { + environment: "test".to_string(), + service_name: "codex-exec-server".to_string(), + service_version: env!("CARGO_PKG_VERSION").to_string(), + codex_home: std::env::current_dir().expect("cwd"), + exporter, + trace_exporter: if export_traces { + OtelExporter::OtlpHttp { + endpoint: format!("{}/v1/traces", collector.uri()), + headers: HashMap::new(), + protocol: OtelHttpProtocol::Json, + tls: None, + } + } else { + OtelExporter::None + }, + metrics_exporter: OtelExporter::None, + runtime_metrics: false, + span_attributes: BTreeMap::new(), + tracestate: BTreeMap::new(), + }) + .expect("OTEL settings") + .expect("OTEL provider"); + let subscriber = tracing_subscriber::registry() + .with(otel.tracing_layer()) + .with(otel.logger_layer()); + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + run(&runtime, &otel, &collector); + }); + runtime + .block_on(otel.shutdown_with_timeout(Duration::from_secs(5))) + .expect("flush OTEL"); + + let mut records = Vec::new(); + for request in runtime + .block_on(collector.received_requests()) + .expect("exported requests") + { + let body: Value = serde_json::from_slice(&request.body).expect("OTLP JSON"); + let Some(resources) = body["resourceLogs"].as_array() else { + continue; + }; + assert!(!String::from_utf8_lossy(&request.body).contains(PRIVATE_PAYLOAD)); + for resource in resources { + for scope in resource["scopeLogs"].as_array().expect("scope logs") { + for record in scope["logRecords"].as_array().expect("log records") { + assert!(record["body"].is_null(), "lifecycle logs have no raw body"); + records.push(record.clone()); + } + } + } + } + records +} + +struct TestServer { + sessions: Arc, + router: RpcRouter, + outgoing: mpsc::Sender, + _notifications: mpsc::Receiver, +} + +impl TestServer { + fn new() -> Self { + let (outgoing, notifications) = mpsc::channel(/*buffer*/ 128); + Self { + sessions: SessionRegistry::new(ExecServerTelemetry::default()), + router: build_router(), + outgoing, + _notifications: notifications, + } + } + + async fn new_orchestrator( + &self, + client_name: &str, + registration: Option<&str>, + ) -> (Arc, String) { + let handler = new_handler(&self.sessions, self.outgoing.clone(), registration); + let initialized = handler + .initialize(InitializeParams { + client_name: client_name.to_string(), + resume_session_id: None, + }) + .await + .expect("initialize orchestrator"); + handler.initialized().expect("initialized orchestrator"); + (handler, initialized.session_id) + } +} + +fn process_start_params(process_id: &str, argv: Value, cwd: PathUri) -> Value { + json!({ + "processId": process_id, + "argv": argv, + "cwd": cwd, + "env": {"PRIVATE_TEST_VALUE": PRIVATE_PAYLOAD}, + "tty": false, + "arg0": null, + }) +} + +fn process_metadata(index: usize) -> Value { + json!({"threadId": THREADS[index], "toolCallId": CALLS[index]}) +} + +fn new_handler( + sessions: &Arc, + outgoing: mpsc::Sender, + registration: Option<&str>, +) -> Arc { + let mut handler = ExecServerHandler::new( + Arc::clone(sessions), + RpcNotificationSender::new(outgoing), + ExecServerRuntimePaths::new( + std::env::current_exe().expect("test executable"), + /*codex_linux_sandbox_exe*/ None, + ) + .expect("runtime paths"), + HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + ); + handler.executor_registration = registration + .and_then(|registration| { + ExecutorRegistration::new("environment".to_string(), registration.to_string()) + }) + .map(Arc::new); + Arc::new(handler) +} diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 19da5f9361..1fd2a1f1d6 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -24,6 +24,7 @@ use crate::server::request_dispatcher::RequestTaskResult; use crate::server::session_registry::SessionRegistry; use crate::telemetry::ConnectionTransport; use crate::telemetry::ExecServerTelemetry; +use crate::telemetry::ExecutorRegistration; use codex_http_client::HttpClientFactory; #[derive(Clone)] @@ -72,12 +73,23 @@ impl ConnectionProcessor { ) { run_connection( connection, - Arc::clone(&self.session_registry), - self.runtime_paths.clone(), - self.telemetry.clone(), - self.http_client_factory.clone(), + self.clone(), transport, - self.request_dispatch_mode, + /*executor_registration*/ None, + ) + .await; + } + + pub(crate) async fn run_registered_connection( + &self, + connection: JsonRpcConnection, + executor_registration: Option>, + ) { + run_connection( + connection, + self.clone(), + ConnectionTransport::Relay, + executor_registration, ) .await; } @@ -89,13 +101,17 @@ impl ConnectionProcessor { async fn run_connection( connection: JsonRpcConnection, - session_registry: Arc, - runtime_paths: ExecServerRuntimePaths, - telemetry: ExecServerTelemetry, - http_client_factory: HttpClientFactory, + processor: ConnectionProcessor, transport: ConnectionTransport, - request_dispatch_mode: RequestDispatchMode, + executor_registration: Option>, ) { + let ConnectionProcessor { + session_registry, + runtime_paths, + telemetry, + http_client_factory, + request_dispatch_mode, + } = processor; let _connection_metrics = telemetry.connection_started(transport); let JsonRpcConnection { outgoing_tx: json_outgoing_tx, @@ -108,12 +124,14 @@ async fn run_connection( mpsc::channel::(CHANNEL_CAPACITY); let notifications = RpcNotificationSender::new(outgoing_tx.clone()); let requests = notifications.request_sender(); - let handler = Arc::new(ExecServerHandler::new( + let mut handler = ExecServerHandler::new( session_registry, notifications, runtime_paths, http_client_factory, - )); + ); + handler.executor_registration = executor_registration; + let handler = Arc::new(handler); let outbound_task = tokio::spawn(async move { while let Some(message) = outgoing_rx.recv().await { @@ -297,7 +315,6 @@ mod tests { use crate::protocol::TerminateResponse; use crate::rpc::RpcServerOutboundMessage; use crate::rpc_server_requests::RpcServerRequestSender; - use crate::server::RequestDispatchMode; use crate::server::session_registry::SessionRegistry; #[tokio::test] @@ -494,14 +511,12 @@ mod tests { JsonRpcConnection::from_stdio(server_reader, server_writer, label.to_string()); let task = tokio::spawn(run_connection( connection, - registry, - test_runtime_paths(), - crate::ExecServerTelemetry::default(), - codex_http_client::HttpClientFactory::new( - codex_http_client::OutboundProxyPolicy::ReqwestDefault, - ), + super::ConnectionProcessor { + session_registry: registry, + ..super::ConnectionProcessor::new(test_runtime_paths()) + }, crate::telemetry::ConnectionTransport::Stdio, - RequestDispatchMode::Inline, + /*executor_registration*/ None, )); (client_writer, BufReader::new(client_reader).lines(), task) } @@ -574,6 +589,7 @@ mod tests { env.insert("PATH".to_string(), path.to_string_lossy().into_owned()); } ExecParams { + metadata: Default::default(), process_id, argv: sleep_then_print_argv(), cwd: PathUri::from_host_native_path(std::env::current_dir().expect("cwd")) diff --git a/codex-rs/exec-server/src/server/registry.rs b/codex-rs/exec-server/src/server/registry.rs index b1a1173085..4cd21ad85e 100644 --- a/codex-rs/exec-server/src/server/registry.rs +++ b/codex-rs/exec-server/src/server/registry.rs @@ -1,3 +1,4 @@ +use opentelemetry::trace::TraceContextExt; use std::sync::Arc; use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD; @@ -68,9 +69,15 @@ pub(crate) fn build_router() -> RpcRouter { handler.http_request(request_id, params).await }, ); - router.request( + router.request_with_trace( EXEC_METHOD, - |handler: Arc, params: ExecParams| async move { handler.exec(params).await }, + |handler: Arc, params: ExecParams, trace| async move { + let launch_context = trace + .as_ref() + .and_then(codex_otel::context_from_w3c_trace_context) + .map(|context| context.span().span_context().clone()); + handler.exec(params, launch_context).await + }, ); router.request( ENVIRONMENT_INFO_METHOD, diff --git a/codex-rs/exec-server/src/shell_snapshot_tests.rs b/codex-rs/exec-server/src/shell_snapshot_tests.rs index d0dcf1f88e..b5df524235 100644 --- a/codex-rs/exec-server/src/shell_snapshot_tests.rs +++ b/codex-rs/exec-server/src/shell_snapshot_tests.rs @@ -15,7 +15,8 @@ use super::MAX_SNAPSHOT_ATTEMPTS; use super::SNAPSHOT_RETRY_BACKOFF; use super::ShellSnapshotCache; use super::parse_snapshot; -use crate::process_sandbox::prepare_exec_request; +use crate::process_sandbox::prepare_exec_request_with_telemetry; +use crate::process_telemetry::ProcessTelemetry; use crate::protocol::ExecEnvPolicy; use crate::protocol::ExecParams; use crate::protocol::ProcessId; @@ -40,6 +41,7 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight( let profile = home.path().join(".bashrc"); std::fs::write(&profile, "printf x >> \"$HOME/captures\"\nexit 7\n")?; let params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("snapshot-retry"), argv: vec![ "/bin/bash".to_string(), @@ -89,21 +91,23 @@ async fn snapshot_failure_retries_are_bounded_and_single_flight( "printf x >> \"$HOME/captures\"\nprofile_helper() { printf recovered; }\n", )?; } - let mut prepared = prepare_exec_request( + let mut prepared = prepare_exec_request_with_telemetry( ¶ms, params.env.clone(), /*runtime_paths*/ None, /*network_policy_decider*/ None, /*network_policy_audit_observer*/ None, + &ProcessTelemetry::default(), ) .await .expect("prepare capture"); - let mut concurrent = prepare_exec_request( + let mut concurrent = prepare_exec_request_with_telemetry( ¶ms, params.env.clone(), /*runtime_paths*/ None, /*network_policy_decider*/ None, /*network_policy_audit_observer*/ None, + &ProcessTelemetry::default(), ) .await .expect("prepare concurrent capture"); diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 725e018eeb..8f54473691 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -6,6 +6,27 @@ use std::time::Instant; use codex_otel::MetricsClient; use tracing::warn; +/// Registry-issued identity captured from the executor's authenticated relay connection. +pub(crate) struct ExecutorRegistration { + pub(crate) environment_id: String, + pub(crate) executor_registration_id: String, +} + +impl ExecutorRegistration { + pub(crate) fn new(environment_id: String, executor_registration_id: String) -> Option { + if [&environment_id, &executor_registration_id] + .iter() + .any(|id| id.trim().is_empty() || id.len() > 256 || id.chars().any(char::is_control)) + { + return None; + } + Some(Self { + environment_id, + executor_registration_id, + }) + } +} + const CONNECTIONS_ACTIVE_METRIC: &str = "exec_server_connections_active"; const CONNECTIONS_ACTIVE_DESCRIPTION: &str = "Number of active exec-server connections."; const CONNECTIONS_TOTAL_METRIC: &str = "exec_server_connections_total"; diff --git a/codex-rs/exec-server/tests/accepted_websocket.rs b/codex-rs/exec-server/tests/accepted_websocket.rs index 5334dfd7fd..8ce621fa1a 100644 --- a/codex-rs/exec-server/tests/accepted_websocket.rs +++ b/codex-rs/exec-server/tests/accepted_websocket.rs @@ -214,6 +214,7 @@ async fn accepted_websocket_reconnect_recovers_running_process_and_output() -> R async move { backend .start(ExecParams { + metadata: Default::default(), process_id, argv: vec!["test-command".to_string()], cwd: PathUri::parse("file:///workspace")?, diff --git a/codex-rs/exec-server/tests/exec_process.rs b/codex-rs/exec-server/tests/exec_process.rs index 82fc6590f5..823f626c51 100644 --- a/codex-rs/exec-server/tests/exec_process.rs +++ b/codex-rs/exec-server/tests/exec_process.rs @@ -227,6 +227,7 @@ async fn shell_snapshot_v2_filters_profile_exports_and_stays_in_memory( let started = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from(format!("snapshot-{attempt}")), argv: vec![shell_path.to_string(), "-lc".to_string(), command.clone()], cwd: cwd.clone(), @@ -303,6 +304,7 @@ async fn shell_snapshot_v2_remote_managed_proxy_uses_prepared_execution_context( let started = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from(format!("managed-snapshot-{attempt}")), argv: vec![ "/bin/bash".to_string(), @@ -397,6 +399,7 @@ async fn shell_snapshot_v2_capture_failure_falls_back_and_retries( include_only: vec!["HOME".to_string(), "PATH".to_string()], }; let mut params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("snapshot-first"), argv: vec![ shell_path.to_string(), @@ -505,6 +508,7 @@ async fn remote_sandboxed_process_preserves_custom_arg0() -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-custom-arg0"), argv: vec![ "/bin/sh".to_string(), @@ -545,6 +549,7 @@ async fn assert_exec_process_starts_and_exits(use_remote: bool) -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-1"), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path(std::env::current_dir()?)?, @@ -607,6 +612,7 @@ async fn remote_process_keeps_sandbox_helper_visible_with_restricted_reads() -> let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-restricted-helper"), argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()], cwd, @@ -677,6 +683,7 @@ async fn remote_tty_process_uses_configured_sandbox_helper_with_hostile_path() - let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-hostile-helper-path"), argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()], cwd, @@ -733,6 +740,7 @@ async fn remote_process_preserves_empty_workspace_roots() -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-empty-workspace-roots"), argv: vec!["/bin/cat".to_string(), file.to_string_lossy().into_owned()], cwd, @@ -875,6 +883,7 @@ async fn assert_exec_process_streams_output(use_remote: bool) -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -911,6 +920,7 @@ async fn assert_exec_process_pushes_events(use_remote: bool) -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -963,6 +973,7 @@ async fn assert_exec_process_replays_events_after_close(use_remote: bool) -> Res let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -1016,6 +1027,7 @@ async fn assert_exec_process_retains_output_after_exit_until_streams_close( let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ helper_binary.to_string_lossy().into_owned(), @@ -1091,6 +1103,7 @@ async fn assert_exec_process_write_then_read(use_remote: bool) -> Result<()> { let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ // Use `/bin/sh` instead of Python so this stdin round-trip test @@ -1136,6 +1149,7 @@ async fn assert_exec_process_write_then_read_without_tty(use_remote: bool) -> Re let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -1182,6 +1196,7 @@ async fn assert_remote_windows_sandbox_process_write() -> Result<()> { let session = match context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-windows-sandbox-stdin"), argv: vec![ r"C:\Windows\System32\cmd.exe".to_string(), @@ -1234,6 +1249,7 @@ async fn assert_exec_process_rejects_write_without_pipe_stdin(use_remote: bool) let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -1273,6 +1289,7 @@ async fn assert_exec_process_signal_interrupts_process(use_remote: bool) -> Resu let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: process_id.clone().into(), argv: vec![ "/bin/sh".to_string(), @@ -1331,6 +1348,7 @@ async fn assert_exec_process_signal_terminates_on_windows(use_remote: bool) -> R let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-windows-signal"), argv: vec![ "cmd".to_string(), @@ -1368,6 +1386,7 @@ async fn assert_exec_process_preserves_queued_events_before_subscribe( let session = context .backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-queued"), argv: vec![ "/bin/sh".to_string(), @@ -1413,6 +1432,7 @@ async fn remote_exec_process_recovers_after_transport_disconnect() -> Result<()> let emitted_path = temp_dir.path().join("output-emitted"); let session = backend .start(ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-recover"), argv: vec![ "/bin/sh".to_string(), diff --git a/codex-rs/exec-server/tests/forward.rs b/codex-rs/exec-server/tests/forward.rs index 02c272bdae..526bd12119 100644 --- a/codex-rs/exec-server/tests/forward.rs +++ b/codex-rs/exec-server/tests/forward.rs @@ -62,6 +62,7 @@ async fn forwarder_runs_commands_and_transfers_files() -> Result<()> { timeout( TEST_TIMEOUT, client.exec(ExecParams { + metadata: Default::default(), process_id: process_id.clone(), argv, cwd: PathUri::from_host_native_path(std::env::current_dir()?)?, diff --git a/codex-rs/exec-server/tests/relay.rs b/codex-rs/exec-server/tests/relay.rs index e4e44c2a28..ae56123bdf 100644 --- a/codex-rs/exec-server/tests/relay.rs +++ b/codex-rs/exec-server/tests/relay.rs @@ -323,6 +323,7 @@ async fn remote_environment_routes_encrypted_exec_server_rpc() -> Result<()> { let client = &connection.client; let exec_params = ExecParams { + metadata: Default::default(), process_id: ProcessId::from("proc-1"), argv: vec!["true".to_string()], cwd: PathUri::from_host_native_path(std::env::current_dir()?)?, diff --git a/codex-rs/network-proxy/Cargo.toml b/codex-rs/network-proxy/Cargo.toml index 82407b00d1..de72a7684e 100644 --- a/codex-rs/network-proxy/Cargo.toml +++ b/codex-rs/network-proxy/Cargo.toml @@ -21,6 +21,7 @@ codex-utils-absolute-path = { workspace = true } codex-utils-home-dir = { workspace = true } codex-utils-rustls-provider = { workspace = true } globset = { workspace = true } +opentelemetry = { workspace = true } rand = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } diff --git a/codex-rs/network-proxy/src/lib.rs b/codex-rs/network-proxy/src/lib.rs index 060ee0cdd6..5ffb23046d 100644 --- a/codex-rs/network-proxy/src/lib.rs +++ b/codex-rs/network-proxy/src/lib.rs @@ -13,6 +13,7 @@ mod mitm_hook; mod native_certs; mod network_policy; mod policy; +mod process_log_metadata; mod proxy; mod reasons; mod remote_config; @@ -63,6 +64,8 @@ pub use network_policy::NetworkPolicyRequest; pub use network_policy::NetworkPolicyRequestArgs; pub use network_policy::NetworkProtocol; pub use policy::normalize_host; +pub use process_log_metadata::ExecutorLogIdentity; +pub use process_log_metadata::NetworkProxyProcessLogMetadata; pub use proxy::ALL_PROXY_ENV_KEYS; pub use proxy::ALLOW_LOCAL_BINDING_ENV_KEY; pub use proxy::Args; diff --git a/codex-rs/network-proxy/src/network_policy.rs b/codex-rs/network-proxy/src/network_policy.rs index b5d2d2719e..1839c371e4 100644 --- a/codex-rs/network-proxy/src/network_policy.rs +++ b/codex-rs/network-proxy/src/network_policy.rs @@ -260,22 +260,41 @@ struct PolicyAuditEventArgs<'a> { } fn emit_policy_audit_event(state: &NetworkProxyState, args: PolicyAuditEventArgs<'_>) { - let metadata = state.audit_metadata(); + let audit_metadata = state.audit_metadata(); + let process_log_metadata = &state.process_log_metadata; + let conversation_id = process_log_metadata + .thread_id + .as_deref() + .or(audit_metadata.conversation_id.as_deref()); let timestamp = audit_timestamp(); + let launch_trace_id = state + .launch_span_context + .as_ref() + .map(|context| context.trace_id().to_string()); + let launch_span_id = state + .launch_span_context + .as_ref() + .map(|context| context.span_id().to_string()); + let executor_identity = process_log_metadata.executor_identity.as_ref(); tracing::event!( target: AUDIT_TARGET, tracing::Level::INFO, event.name = POLICY_DECISION_EVENT_NAME, event.timestamp = %timestamp, - conversation.id = metadata.conversation_id.as_deref(), - app.version = metadata.app_version.as_deref(), - auth_mode = metadata.auth_mode.as_deref(), - originator = metadata.originator.as_deref(), - user.account_id = metadata.user_account_id.as_deref(), - user.email = metadata.user_email.as_deref(), - terminal.type = metadata.terminal_type.as_deref(), - model = metadata.model.as_deref(), - slug = metadata.slug.as_deref(), + launch.trace_id = launch_trace_id.as_deref(), + launch.span_id = launch_span_id.as_deref(), + conversation.id = conversation_id, + tool.call_id = process_log_metadata.tool_call_id.as_deref(), + executor.environment_id = executor_identity.map(|identity| identity.environment_id.as_str()), + executor.registration_id = executor_identity.map(|identity| identity.registration_id.as_str()), + app.version = audit_metadata.app_version.as_deref(), + auth_mode = audit_metadata.auth_mode.as_deref(), + originator = audit_metadata.originator.as_deref(), + user.account_id = audit_metadata.user_account_id.as_deref(), + user.email = audit_metadata.user_email.as_deref(), + terminal.type = audit_metadata.terminal_type.as_deref(), + model = audit_metadata.model.as_deref(), + slug = audit_metadata.slug.as_deref(), network.policy.scope = args.scope, network.policy.decision = args.decision, network.policy.source = args.source, @@ -601,6 +620,8 @@ mod tests { use super::test_support::capture_events; use super::test_support::find_event_by_name; use super::*; + use crate::ExecutorLogIdentity; + use crate::NetworkProxyProcessLogMetadata; use crate::config::NetworkMode; use crate::config::NetworkProxyConfig; use crate::reasons::REASON_DENIED; @@ -937,7 +958,7 @@ mod tests { model: Some("gpt-5.3-codex".to_string()), slug: Some("gpt-5.3-codex".to_string()), }; - let state = state_with_metadata(metadata); + let mut state = state_with_metadata(metadata); let request = NetworkPolicyRequest::new(NetworkPolicyRequestArgs { protocol: NetworkProtocol::Http, host: "example.com".to_string(), @@ -949,24 +970,50 @@ mod tests { exec_policy_hint: None, }); - let (_decision, events) = capture_events(|| async { - evaluate_host_policy(&state, /*decider*/ None, &request) - .await - .unwrap() - }) - .await; + for (thread_id, expected_conversation_id) in [ + (None, "conversation-1"), + (Some("process-thread-1"), "process-thread-1"), + (None, "conversation-1"), + ] { + state.set_process_log_metadata(NetworkProxyProcessLogMetadata { + thread_id: thread_id.map(str::to_string), + tool_call_id: Some("call-1".to_string()), + executor_identity: Some(ExecutorLogIdentity { + environment_id: "environment-1".to_string(), + registration_id: "registration-1".to_string(), + }), + }); + let (_decision, events) = capture_events(|| async { + evaluate_host_policy(&state, /*decider*/ None, &request) + .await + .unwrap() + }) + .await; - let event = find_event_by_name(&events, POLICY_DECISION_EVENT_NAME) - .expect("expected policy decision audit event"); - assert_eq!(event.field("conversation.id"), Some("conversation-1")); - assert_eq!(event.field("app.version"), Some("1.2.3")); - assert_eq!(event.field("auth_mode"), Some("Chatgpt")); - assert_eq!(event.field("originator"), Some("codex_cli_rs")); - assert_eq!(event.field("user.account_id"), Some("acct-1")); - assert_eq!(event.field("user.email"), Some("test@example.com")); - assert_eq!(event.field("terminal.type"), Some("iTerm.app/3.6.5")); - assert_eq!(event.field("model"), Some("gpt-5.3-codex")); - assert_eq!(event.field("slug"), Some("gpt-5.3-codex")); + let event = find_event_by_name(&events, POLICY_DECISION_EVENT_NAME) + .expect("expected policy decision audit event"); + assert_eq!( + event.field("conversation.id"), + Some(expected_conversation_id) + ); + assert_eq!(event.field("tool.call_id"), Some("call-1")); + assert_eq!( + event.field("executor.environment_id"), + Some("environment-1") + ); + assert_eq!( + event.field("executor.registration_id"), + Some("registration-1") + ); + assert_eq!(event.field("app.version"), Some("1.2.3")); + assert_eq!(event.field("auth_mode"), Some("Chatgpt")); + assert_eq!(event.field("originator"), Some("codex_cli_rs")); + assert_eq!(event.field("user.account_id"), Some("acct-1")); + assert_eq!(event.field("user.email"), Some("test@example.com")); + assert_eq!(event.field("terminal.type"), Some("iTerm.app/3.6.5")); + assert_eq!(event.field("model"), Some("gpt-5.3-codex")); + assert_eq!(event.field("slug"), Some("gpt-5.3-codex")); + } } #[tokio::test(flavor = "current_thread")] diff --git a/codex-rs/network-proxy/src/process_log_metadata.rs b/codex-rs/network-proxy/src/process_log_metadata.rs new file mode 100644 index 0000000000..cab758f45d --- /dev/null +++ b/codex-rs/network-proxy/src/process_log_metadata.rs @@ -0,0 +1,17 @@ +//! Process attribution captured at launch for network policy logs. +//! Executor registration identity comes from the executor, never the controller. + +/// Registry-issued identity of the executor that launched a process. +#[derive(Clone)] +pub struct ExecutorLogIdentity { + pub environment_id: String, + pub registration_id: String, +} + +/// Correlation metadata only; these values do not authorize network requests. +#[derive(Clone, Default)] +pub struct NetworkProxyProcessLogMetadata { + pub thread_id: Option, + pub tool_call_id: Option, + pub executor_identity: Option, +} diff --git a/codex-rs/network-proxy/src/runtime.rs b/codex-rs/network-proxy/src/runtime.rs index 1889f827a1..9c9f8c6fc9 100644 --- a/codex-rs/network-proxy/src/runtime.rs +++ b/codex-rs/network-proxy/src/runtime.rs @@ -24,6 +24,7 @@ use anyhow::Context; use anyhow::Result; use codex_utils_absolute_path::AbsolutePathBuf; use globset::GlobSet; +use opentelemetry::trace::SpanContext; use serde::Deserialize; use serde::Serialize; use std::collections::HashMap; @@ -234,6 +235,8 @@ pub struct NetworkProxyState { reloader: Arc, blocked_request_observer: Arc>>>, pub(crate) policy_audit_observer: Option, + pub(crate) launch_span_context: Option, + pub(crate) process_log_metadata: crate::NetworkProxyProcessLogMetadata, credential_broker: CredentialBroker, audit_metadata: NetworkProxyAuditMetadata, execution_attributions: Arc>>, @@ -269,6 +272,8 @@ impl Clone for NetworkProxyState { reloader: self.reloader.clone(), blocked_request_observer: self.blocked_request_observer.clone(), policy_audit_observer: self.policy_audit_observer.clone(), + launch_span_context: self.launch_span_context.clone(), + process_log_metadata: self.process_log_metadata.clone(), credential_broker: self.credential_broker.clone(), audit_metadata: self.audit_metadata.clone(), execution_attributions: self.execution_attributions.clone(), @@ -355,6 +360,8 @@ impl NetworkProxyState { reloader, blocked_request_observer: Arc::new(RwLock::new(blocked_request_observer)), policy_audit_observer: None, + launch_span_context: None, + process_log_metadata: crate::NetworkProxyProcessLogMetadata::default(), audit_metadata, execution_attributions: Arc::new(Mutex::new(HashMap::new())), environment_id: None, @@ -422,6 +429,16 @@ impl NetworkProxyState { self.policy_audit_observer = Some(observer); } + /// Retains the initiating process request's trace reference for network decision logs. + pub fn set_launch_span_context(&mut self, span_context: SpanContext) { + self.launch_span_context = span_context.is_valid().then_some(span_context); + } + + /// Adds launch-local log fields. Registration IDs must come from the executor, not the peer. + pub fn set_process_log_metadata(&mut self, metadata: crate::NetworkProxyProcessLogMetadata) { + self.process_log_metadata = metadata; + } + pub fn audit_metadata(&self) -> &NetworkProxyAuditMetadata { &self.audit_metadata } diff --git a/codex-rs/otel/src/trace_context.rs b/codex-rs/otel/src/trace_context.rs index 6ae924d1d8..8ce0f99a4b 100644 --- a/codex-rs/otel/src/trace_context.rs +++ b/codex-rs/otel/src/trace_context.rs @@ -137,7 +137,8 @@ pub(crate) fn context_from_trace_headers( headers.insert("tracestate".to_string(), tracestate.to_string()); } - let context = TraceContextPropagator::new().extract(&headers); + // Invalid incoming headers must not fall back to an unrelated active span. + let context = TraceContextPropagator::new().extract_with_context(&Context::new(), &headers); if !context.span().span_context().is_valid() { return None; } @@ -369,6 +370,12 @@ mod tests { #[test] fn invalid_traceparent_returns_none() { + let parent = context_from_trace_headers( + Some("00-11111111111111111111111111111111-2222222222222222-01"), + /*tracestate*/ None, + ) + .expect("valid ambient context"); + let _guard = parent.attach(); assert!( context_from_trace_headers(Some("not-a-traceparent"), /*tracestate*/ None).is_none() ); diff --git a/codex-rs/rmcp-client/src/stdio_server_launcher.rs b/codex-rs/rmcp-client/src/stdio_server_launcher.rs index 7c636e3f3d..f942378bb8 100644 --- a/codex-rs/rmcp-client/src/stdio_server_launcher.rs +++ b/codex-rs/rmcp-client/src/stdio_server_launcher.rs @@ -578,6 +578,7 @@ impl ExecutorStdioServerLauncher { // rmcp write JSON-RPC requests after the process starts. let started = exec_backend .start(ExecParams { + metadata: Default::default(), process_id, argv, cwd,