diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index edbdcb379f..dcb05293fb 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -45,6 +45,7 @@ use crate::environment::EnvironmentConnectionState; use crate::process::ExecProcessEvent; use crate::process::ExecProcessEventLog; use crate::process::ExecProcessEventReceiver; +use crate::process_telemetry::trace_process_id; use crate::protocol::CAPABILITY_ROOTS_DISCOVER_METHOD; use crate::protocol::CapabilityRootsDiscoverParams; use crate::protocol::CapabilityRootsDiscoverResponse; @@ -1026,6 +1027,11 @@ impl ExecServerClient { .await } + #[tracing::instrument( + name = "codex.exec_server.process_start", + skip_all, + fields(process.id = trace_process_id(params.process_id.as_str())), + )] pub(crate) async fn start_process( &self, params: ExecParams, diff --git a/codex-rs/exec-server/src/connection.rs b/codex-rs/exec-server/src/connection.rs index a0e4a18d86..476aaf3619 100644 --- a/codex-rs/exec-server/src/connection.rs +++ b/codex-rs/exec-server/src/connection.rs @@ -68,11 +68,14 @@ impl JsonRpcConnectionEvent { }; let queued_at = Instant::now(); + // Record phase offsets from receipt because detached work can keep the span open. let request_span = tracing::info_span!( "codex.exec_server.request", otel.kind = "server", otel.name = "unknown", method = request.method.as_str(), + rpc.dispatch_offset_ns = tracing::field::Empty, + rpc.response_enqueue_offset_ns = tracing::field::Empty, result = tracing::field::Empty, ); if let Some(trace) = &request.trace diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index e96348ff77..ddee602bb4 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -36,6 +36,7 @@ mod remote_process; mod resolved_capability; mod rpc; mod rpc_server_requests; +mod rpc_timing; mod runtime_paths; mod sandbox_selection; mod sandboxed_file_open; diff --git a/codex-rs/exec-server/src/process_sandbox.rs b/codex-rs/exec-server/src/process_sandbox.rs index ada35f738a..a1a43323ba 100644 --- a/codex-rs/exec-server/src/process_sandbox.rs +++ b/codex-rs/exec-server/src/process_sandbox.rs @@ -36,6 +36,7 @@ use codex_utils_path_uri::PathUri; #[cfg(unix)] use crate::CODEX_ARG0_EXEC_HELPER_ARG1; use crate::ExecServerRuntimePaths; +use crate::process_telemetry::trace_process_id; use crate::protocol::ExecParams; use crate::rpc::internal_error; use crate::rpc::invalid_params; @@ -77,6 +78,11 @@ impl PreparedExecRequest { } } +#[tracing::instrument( + name = "codex.exec_server.process_prepare_sandbox", + skip_all, + fields(process.id = trace_process_id(params.process_id.as_str())), +)] pub(crate) async fn prepare_exec_request_with_telemetry( params: &ExecParams, env: HashMap, diff --git a/codex-rs/exec-server/src/process_telemetry.rs b/codex-rs/exec-server/src/process_telemetry.rs index f926fa9889..89eefee40f 100644 --- a/codex-rs/exec-server/src/process_telemetry.rs +++ b/codex-rs/exec-server/src/process_telemetry.rs @@ -1,5 +1,5 @@ -//! Emits bounded process lifecycle telemetry using identity captured at launch. -//! Reconnects and later process operations must not replace that identity. +//! Emits process lifecycle telemetry and bounds process identifiers recorded by tracing. +//! Lifecycle events retain the identity captured at launch across reconnects and later operations. use std::sync::Arc; @@ -76,3 +76,7 @@ impl ProcessTelemetry { ); } } + +pub(crate) fn trace_process_id(process_id: &str) -> Option<&str> { + (!process_id.is_empty() && process_id.len() <= 64).then_some(process_id) +} diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 5047b06562..35fe69468f 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -36,6 +36,7 @@ use crate::connection::JsonRpcConnection; use crate::connection::JsonRpcConnectionEvent; use crate::connection::JsonRpcTransport; use crate::rpc_server_requests::RpcServerRequestSender; +use crate::rpc_timing::RpcCompletion; #[cfg(test)] #[path = "rpc_client_metrics_tests.rs"] @@ -60,7 +61,7 @@ pub(crate) enum RpcCallError { PendingRequestLimitExceeded { limit: usize }, } -type PendingRequest = oneshot::Sender>; +type PendingRequest = oneshot::Sender; type BoxFuture = Pin + Send + 'static>>; type RequestRoute = Box< dyn Fn(Arc, JSONRPCRequest) -> BoxFuture> + Send + Sync, @@ -666,7 +667,13 @@ impl RpcClient { self.pending.lock().await.remove(&request_id); return Err(RpcCallError::Closed); } - + tracing::event!( + name: "codex.exec_server.request_enqueued", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.exec_server.request_enqueued", + rpc.method = method, + ); // Do not race in-flight requests directly against the transport-close // watch value. The connection reader receives JSON-RPC messages and // the terminal disconnect event on one ordered queue, then drains any @@ -686,11 +693,9 @@ impl RpcClient { } }, }; - let result: Result = response.map_err(|_| RpcCallError::Closed)?; - let response = match result { - Ok(response) => response, - Err(error) => return Err(error), - }; + let completion = response.map_err(|_| RpcCallError::Closed)?; + completion.record_receipt(method); + let response = completion.result?; serde_json::from_value(response).map_err(RpcCallError::Json) } @@ -821,12 +826,12 @@ async fn handle_server_message( match message { JSONRPCMessage::Response(JSONRPCResponse { id, result }) => { if let Some(pending) = pending.lock().await.remove(&id) { - let _ = pending.send(Ok(result)); + let _ = pending.send(RpcCompletion::new(Ok(result))); } } JSONRPCMessage::Error(JSONRPCError { id, error }) => { if let Some(pending) = pending.lock().await.remove(&id) { - let _ = pending.send(Err(RpcCallError::Server(error))); + let _ = pending.send(RpcCompletion::new(Err(RpcCallError::Server(error)))); } } JSONRPCMessage::Notification(notification) => { @@ -857,7 +862,7 @@ async fn drain_pending(pending: &Mutex>) { .collect::>() }; for pending in pending { - let _ = pending.send(Err(RpcCallError::Closed)); + let _ = pending.send(RpcCompletion::new(Err(RpcCallError::Closed))); } } diff --git a/codex-rs/exec-server/src/rpc_timing.rs b/codex-rs/exec-server/src/rpc_timing.rs new file mode 100644 index 0000000000..6375d0b815 --- /dev/null +++ b/codex-rs/exec-server/src/rpc_timing.rs @@ -0,0 +1,44 @@ +//! Carries RPC reader completion time back to the caller so tracing stays in the awaiting task. + +use std::time::Instant; + +use serde_json::Value; + +use crate::rpc::RpcCallError; + +pub(crate) struct RpcCompletion { + pub(crate) result: Result, + completed_at: Instant, +} + +impl RpcCompletion { + pub(crate) fn new(result: Result) -> Self { + Self { + result, + completed_at: Instant::now(), + } + } + + pub(crate) fn record_receipt(&self, method: &str) { + let outcome = match &self.result { + Ok(_) => "success", + Err(RpcCallError::Server(_)) => "error", + Err( + RpcCallError::Closed + | RpcCallError::Json(_) + | RpcCallError::TimedOut { .. } + | RpcCallError::PendingRequestLimitExceeded { .. }, + ) => return, + }; + tracing::event!( + name: "codex.exec_server.response_received", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.exec_server.response_received", + rpc.method = method, + outcome, + reader_to_caller_ns = i64::try_from(self.completed_at.elapsed().as_nanos()) + .unwrap_or(i64::MAX), + ); + } +} diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 3041a102a7..906cc681f7 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -52,6 +52,7 @@ use crate::protocol::FsWalkParams; use crate::protocol::FsWalkResponse; use crate::protocol::FsWriteFileParams; use crate::protocol::FsWriteFileResponse; +use crate::protocol::HTTP_REQUEST_METHOD; use crate::protocol::HttpRequestParams; use crate::protocol::InitializeParams; use crate::protocol::InitializeResponse; @@ -292,6 +293,15 @@ impl ExecServerHandler { } return Err(error); } + // This response bypasses the dispatcher; record it before body-stream setup. + tracing::event!( + name: "codex.exec_server.response_enqueued", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.exec_server.response_enqueued", + rpc.method = HTTP_REQUEST_METHOD, + outcome = "success", + ); if let Some(pending_stream) = pending_stream { self.start_http_body_stream(pending_stream).await; } diff --git a/codex-rs/exec-server/src/server/request_dispatcher.rs b/codex-rs/exec-server/src/server/request_dispatcher.rs index c1f16b64fe..cc78c248c6 100644 --- a/codex-rs/exec-server/src/server/request_dispatcher.rs +++ b/codex-rs/exec-server/src/server/request_dispatcher.rs @@ -182,6 +182,10 @@ impl RequestDispatcher { self.telemetry .request_queue_completed(method, queued_at.elapsed()); request_span.record("otel.name", method); + request_span.record( + "rpc.dispatch_offset_ns", + i64::try_from(queued_at.elapsed().as_nanos()).unwrap_or(i64::MAX), + ); if self .outgoing_tx .send(RpcServerOutboundMessage::Error { @@ -203,6 +207,10 @@ impl RequestDispatcher { ); return RequestTaskResult::ConnectionClosed; } + request_span.record( + "rpc.response_enqueue_offset_ns", + i64::try_from(queued_at.elapsed().as_nanos()).unwrap_or(i64::MAX), + ); request_span.record("result", "error"); self.telemetry.request_completed( method, @@ -221,6 +229,10 @@ impl RequestDispatcher { let mut disconnected_rx = self.disconnected_rx.clone(); let telemetry = self.telemetry.clone(); let task = async move { + request_span.record( + "rpc.dispatch_offset_ns", + i64::try_from(queued_at.elapsed().as_nanos()).unwrap_or(i64::MAX), + ); telemetry.request_queue_completed( method, queued_at.elapsed().saturating_sub(route_setup_duration), @@ -240,10 +252,19 @@ impl RequestDispatcher { }; let result = request_result(&message); let response_sent = match message { - Some(message) => tokio::select! { - result = outgoing_tx.send(message) => result.is_ok(), - _ = disconnected_rx.changed() => false, - }, + Some(message) => { + let response_sent = tokio::select! { + result = outgoing_tx.send(message) => result.is_ok(), + _ = disconnected_rx.changed() => false, + }; + if response_sent { + request_span.record( + "rpc.response_enqueue_offset_ns", + i64::try_from(queued_at.elapsed().as_nanos()).unwrap_or(i64::MAX), + ); + } + response_sent + } None => true, }; if !response_sent { diff --git a/codex-rs/exec-server/src/shell_snapshot.rs b/codex-rs/exec-server/src/shell_snapshot.rs index e848f6d82c..0ba65b9dc7 100644 --- a/codex-rs/exec-server/src/shell_snapshot.rs +++ b/codex-rs/exec-server/src/shell_snapshot.rs @@ -71,6 +71,14 @@ pub(crate) enum CapturePurpose { } impl ShellSnapshotCache { + #[tracing::instrument( + name = "codex.exec_server.process_prepare_shell_snapshot", + skip_all, + fields( + process.id = crate::process_telemetry::trace_process_id(params.process_id.as_str()), + snapshot_requested = params.shell_snapshot.is_some(), + ), + )] pub(crate) async fn prepare( &self, params: &ExecParams, @@ -255,6 +263,11 @@ impl ShellSnapshotCache { } } +#[tracing::instrument( + name = "codex.exec_server.process_capture_shell_snapshot", + skip_all, + fields(process.id = crate::process_telemetry::trace_process_id(params.process_id.as_str())), +)] async fn capture_snapshot( params: &ExecParams, prepared: &PreparedExecRequest, diff --git a/codex-rs/sandboxing/src/spawn.rs b/codex-rs/sandboxing/src/spawn.rs index 053d3c4b56..8bc7447766 100644 --- a/codex-rs/sandboxing/src/spawn.rs +++ b/codex-rs/sandboxing/src/spawn.rs @@ -41,6 +41,7 @@ pub struct SpawnRequest<'a> { } /// Spawn a process using the backend selected by the prepared sandbox request. +#[tracing::instrument(name = "codex.process.spawn", skip_all)] pub async fn spawn_process(request: SpawnRequest<'_>) -> Result { let tty = request.tty; let finish_spawn = |spawned| {