From 3b6a3e700464737aaa1a9d181a8518863d3154d8 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 01:39:58 +0000 Subject: [PATCH 1/3] exec-server: align JSON-RPC span attributes --- codex-rs/exec-server/src/rpc.rs | 32 +++++++++++++++++--- codex-rs/exec-server/src/server/processor.rs | 15 +++++++-- codex-rs/exec-server/src/telemetry.rs | 2 +- 3 files changed, 40 insertions(+), 9 deletions(-) diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 9671f25d3933..72483f22c054 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -363,6 +363,10 @@ impl RpcClient { }) } + fn allocate_request_id(&self) -> RequestId { + RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::SeqCst)) + } + #[tracing::instrument( name = "codex.exec_server.request", level = "info", @@ -370,6 +374,9 @@ impl RpcClient { fields( otel.kind = "client", otel.name = method, + rpc.system = "jsonrpc", + rpc.method = method, + rpc.request_id = tracing::field::Empty, method, ) )] @@ -379,7 +386,10 @@ impl RpcClient { T: DeserializeOwned, { let _call_slot = self.acquire_regular_call_slot()?; - self.call_inner(method, params, RpcCallTimeout::None).await + let request_id = self.allocate_request_id(); + tracing::Span::current().record("rpc.request_id", tracing::field::display(&request_id)); + self.call_inner(request_id, method, params, RpcCallTimeout::None) + .await } pub(crate) async fn call_with_timeout( @@ -393,8 +403,14 @@ impl RpcClient { T: DeserializeOwned, { let _call_slot = self.acquire_regular_call_slot()?; - self.call_inner(method, params, RpcCallTimeout::After(call_timeout)) - .await + let request_id = self.allocate_request_id(); + self.call_inner( + request_id, + method, + params, + RpcCallTimeout::After(call_timeout), + ) + .await } #[tracing::instrument( @@ -404,6 +420,9 @@ impl RpcClient { fields( otel.kind = "client", otel.name = method, + rpc.system = "jsonrpc", + rpc.method = method, + rpc.request_id = tracing::field::Empty, method, ) )] @@ -426,11 +445,15 @@ impl RpcClient { } }, }; - self.call_inner(method, params, RpcCallTimeout::None).await + let request_id = self.allocate_request_id(); + tracing::Span::current().record("rpc.request_id", tracing::field::display(&request_id)); + self.call_inner(request_id, method, params, RpcCallTimeout::None) + .await } async fn call_inner( &self, + request_id: RequestId, method: &str, params: &P, call_timeout: RpcCallTimeout, @@ -439,7 +462,6 @@ impl RpcClient { P: Serialize, T: DeserializeOwned, { - let request_id = RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::SeqCst)); let (response_tx, response_rx) = oneshot::channel(); { let mut pending = self.pending.lock().await; diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 63f6bd45d9a6..9071bd07bef9 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -129,7 +129,7 @@ async fn run_connection( codex_exec_server_protocol::JSONRPCMessage::Request(request) => { let request_started_at = Instant::now(); if let Some((method, route)) = router.request_route(request.method.as_str()) { - let request_span = request_span(method, &request); + let request_span = request_span(method, &request, transport); let message = tokio::select! { message = route(Arc::clone(&handler), request).instrument(request_span.clone()) => message, _ = disconnected_rx.changed() => { @@ -160,7 +160,7 @@ async fn run_connection( drop(request_span); } else { let method = "unknown"; - let request_span = request_span(method, &request); + let request_span = request_span(method, &request, transport); if outgoing_tx .send(RpcServerOutboundMessage::Error { request_id: request.id, @@ -244,12 +244,17 @@ async fn run_connection( fn request_span( span_name: &str, request: &codex_exec_server_protocol::JSONRPCRequest, + transport: ConnectionTransport, ) -> tracing::Span { let method = request.method.as_str(); let span = tracing::info_span!( "codex.exec_server.request", otel.kind = "server", otel.name = span_name, + rpc.system = "jsonrpc", + rpc.method = method, + rpc.transport = transport.metric_tag(), + rpc.request_id = %request.id, method, result = tracing::field::Empty, ); @@ -351,7 +356,11 @@ mod tests { params: None, trace: Some(trace), }; - let request_span = request_span("unknown", &request); + let request_span = request_span( + "unknown", + &request, + crate::telemetry::ConnectionTransport::Stdio, + ); request_span.in_scope(|| {}); drop(request_span); }); diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 650faacb2c63..6e83bb1d5de5 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -52,7 +52,7 @@ pub(crate) enum ConnectionTransport { } impl ConnectionTransport { - fn metric_tag(self) -> &'static str { + pub(crate) fn metric_tag(self) -> &'static str { match self { Self::Relay => "relay", Self::Stdio => "stdio", From 1b4a71eecfa6bb9f9964e58051498d6b7aeee890 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 01:41:04 +0000 Subject: [PATCH 2/3] exec-server: centralize client RPC spans --- codex-rs/exec-server/src/rpc.rs | 41 +++++++++++---------------------- 1 file changed, 13 insertions(+), 28 deletions(-) diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index 72483f22c054..fdb5466c7b01 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -367,19 +367,6 @@ impl RpcClient { RequestId::Integer(self.next_request_id.fetch_add(1, Ordering::SeqCst)) } - #[tracing::instrument( - name = "codex.exec_server.request", - level = "info", - skip_all, - fields( - otel.kind = "client", - otel.name = method, - rpc.system = "jsonrpc", - rpc.method = method, - rpc.request_id = tracing::field::Empty, - method, - ) - )] pub(crate) async fn call(&self, method: &str, params: &P) -> Result where P: Serialize, @@ -387,7 +374,6 @@ impl RpcClient { { let _call_slot = self.acquire_regular_call_slot()?; let request_id = self.allocate_request_id(); - tracing::Span::current().record("rpc.request_id", tracing::field::display(&request_id)); self.call_inner(request_id, method, params, RpcCallTimeout::None) .await } @@ -413,19 +399,6 @@ impl RpcClient { .await } - #[tracing::instrument( - name = "codex.exec_server.request", - level = "info", - skip_all, - fields( - otel.kind = "client", - otel.name = method, - rpc.system = "jsonrpc", - rpc.method = method, - rpc.request_id = tracing::field::Empty, - method, - ) - )] pub(crate) async fn call_for_cleanup( &self, method: &str, @@ -446,11 +419,23 @@ impl RpcClient { }, }; let request_id = self.allocate_request_id(); - tracing::Span::current().record("rpc.request_id", tracing::field::display(&request_id)); self.call_inner(request_id, method, params, RpcCallTimeout::None) .await } + #[tracing::instrument( + name = "codex.exec_server.request", + level = "info", + skip_all, + fields( + otel.kind = "client", + otel.name = method, + rpc.system = "jsonrpc", + rpc.method = method, + rpc.request_id = %request_id, + method, + ) + )] async fn call_inner( &self, request_id: RequestId, From 13f2ede9605b1e1d4e56c79b95684b6b314fd984 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 02:12:18 +0000 Subject: [PATCH 3/3] codex: fix CI failure on PR #31689 --- codex-rs/exec-server/src/client.rs | 1 + codex-rs/exec-server/src/remote.rs | 1 + codex-rs/exec-server/src/rpc.rs | 1 + codex-rs/exec-server/src/server.rs | 1 + codex-rs/exec-server/src/server/processor.rs | 1 + codex-rs/exec-server/src/trace_context_tests.rs | 1 + 6 files changed, 6 insertions(+) diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 2e2f4ce450e1..94992bf587e5 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -1310,6 +1310,7 @@ mod tests { } #[tokio::test(flavor = "current_thread")] + #[serial_test::serial(exec_server_tracing)] async fn process_start_propagates_caller_trace_context_across_background_task() { let (client_stdin, server_reader) = duplex(1 << 20); let (mut server_writer, client_stdout) = duplex(1 << 20); diff --git a/codex-rs/exec-server/src/remote.rs b/codex-rs/exec-server/src/remote.rs index 8870e0efe98c..ef200885a76b 100644 --- a/codex-rs/exec-server/src/remote.rs +++ b/codex-rs/exec-server/src/remote.rs @@ -711,6 +711,7 @@ mod tests { } #[tokio::test(flavor = "current_thread")] + #[serial_test::serial(exec_server_tracing)] async fn register_environment_posts_with_auth_provider_headers() { let provider = SdkTracerProvider::builder().build(); let tracer = provider.tracer("exec-server-test"); diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index fdb5466c7b01..2875efcdfb92 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -974,6 +974,7 @@ mod tests { } #[tokio::test(flavor = "current_thread")] + #[serial_test::serial(exec_server_tracing)] async fn rpc_client_propagates_current_trace_context() { let span_exporter = InMemorySpanExporter::default(); let tracer_provider = SdkTracerProvider::builder() diff --git a/codex-rs/exec-server/src/server.rs b/codex-rs/exec-server/src/server.rs index 7988e0870b6b..39f80c3acf33 100644 --- a/codex-rs/exec-server/src/server.rs +++ b/codex-rs/exec-server/src/server.rs @@ -47,6 +47,7 @@ mod tests { use crate::ExecServerTelemetry; #[tokio::test] + #[serial_test::serial(exec_server_tracing)] async fn telemetry_entrypoint_emits_root_span() { let exporter = InMemorySpanExporter::default(); let provider = SdkTracerProvider::builder() diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 9071bd07bef9..15343754c9ee 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -329,6 +329,7 @@ mod tests { use crate::server::session_registry::SessionRegistry; #[test] + #[serial_test::serial(exec_server_tracing)] fn request_span_uses_bounded_name_wire_method_and_inbound_trace_parent() { let span_exporter = InMemorySpanExporter::default(); let tracer_provider = SdkTracerProvider::builder() diff --git a/codex-rs/exec-server/src/trace_context_tests.rs b/codex-rs/exec-server/src/trace_context_tests.rs index 92a820afc43a..91f3edd8ff72 100644 --- a/codex-rs/exec-server/src/trace_context_tests.rs +++ b/codex-rs/exec-server/src/trace_context_tests.rs @@ -5,6 +5,7 @@ use tracing_subscriber::prelude::*; use super::current_trace_context_headers; #[test] +#[serial_test::serial(exec_server_tracing)] fn creates_traceparent_header_from_current_span() { let provider = SdkTracerProvider::builder().build(); let tracer = provider.tracer("exec-server-test");