Add exec-server RPC timing and process startup tracing (#47680)

## Why

Detached work can keep request spans open beyond response enqueueing. Record explicit phase timings to distinguish request handling from the full span lifetime.

## What changed

- Record server dispatch and response enqueue offsets from request receipt, plus a response enqueue event for HTTP responses that bypass the dispatcher.
- Emit client request enqueue and response receipt events, including success or server-error outcomes and reader-to-caller delay. Carry completion timestamps to the awaiting task so receipt events retain its tracing context.
- Add spans for process startup, sandbox preparation, shell snapshot preparation and capture, and process spawning. Record process identifiers only when nonempty and at most 64 bytes long.

GitOrigin-RevId: 237db898b3284fe8a6a36cb0055b3c7991d8e8a4

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,

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

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;

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<String, String>,

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)
+}

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<Result<Value, RpcCallError>>;
+type PendingRequest = oneshot::Sender<RpcCompletion>;
 type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
 type RequestRoute<S> = Box<
     dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<Option<RpcServerOutboundMessage>> + 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<Value, RpcCallError> = 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<HashMap<RequestId, PendingRequest>>) {
             .collect::<Vec<_>>()
     };
     for pending in pending {
-        let _ = pending.send(Err(RpcCallError::Closed));
+        let _ = pending.send(RpcCompletion::new(Err(RpcCallError::Closed)));
     }
 }
 

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<Value, RpcCallError>,
+    completed_at: Instant,
+}
+
+impl RpcCompletion {
+    pub(crate) fn new(result: Result<Value, RpcCallError>) -> 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),
+        );
+    }
+}

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;
         }

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 {

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,