core: trace executor skill discovery (#30318)

## Why

Make it easier to measure the performance of different parts of skill
loading.

## What

- Add spans for step-context capture, world-state construction, executor
catalog snapshot/root loading, and environment skill loading.
- Record the discovered environment skill count.
- Trace outbound exec-server requests with client kind and RPC method
fields.
- Update trace propagation tests to assert that requests keep the parent
trace id while creating their own child span.

codex-rs/core-skills/src/loader/environment.rs

@@ -143,6 +143,12 @@ pub struct EnvironmentSkillLoadOutcome {
 }
 
 /// Discovers skills without converting environment-owned paths to host paths.
+#[tracing::instrument(
+    name = "skills.environment.load",
+    level = "info",
+    skip_all,
+    fields(skill_count = tracing::field::Empty)
+)]
 pub async fn load_environment_skills_from_root(
     file_system: &dyn ExecutorFileSystem,
     root: &PathUri,
@@ -235,6 +241,7 @@ pub async fn load_environment_skills_from_root(
             warnings: vec![format!("failed to walk skills root {root}: {error:#}")],
         },
     };
+    tracing::Span::current().record("skill_count", discovery.skills.len());
     outcome.warnings.extend(discovery.warnings);
     if discovery.skills.is_empty() {
         return outcome;

codex-rs/core/src/session/mod.rs

@@ -2829,6 +2829,7 @@ impl Session {
     /// This may refresh filesystem-derived state. Normal turns should call it only from
     /// `run_turn` and pass the result down; standalone request or history boundaries may capture
     /// their own step.
+    #[tracing::instrument(name = "step_context.capture", level = "info", skip_all)]
     pub(crate) async fn capture_step_context(
         self: &Arc<Self>,
         turn_context: Arc<TurnContext>,

codex-rs/core/src/session/world_state.rs

@@ -7,6 +7,7 @@ use crate::context::world_state::WorldState;
 use codex_extension_api::WorldStateContributionInput;
 
 impl Session {
+    #[tracing::instrument(name = "world_state.build", level = "info", skip_all)]
     pub(crate) async fn build_world_state_for_step(
         &self,
         step_context: &StepContext,

codex-rs/exec-server/src/client.rs

@@ -1394,7 +1394,16 @@ mod tests {
 
         assert_eq!(session.process_id(), &process_id);
         let trace = server.await.expect("server task").expect("trace context");
-        assert_eq!(trace, expected_trace);
+        let expected_traceparent = expected_trace
+            .traceparent
+            .as_deref()
+            .expect("parent traceparent");
+        let traceparent = trace.traceparent.as_deref().expect("request traceparent");
+        let expected_parts = expected_traceparent.split('-').collect::<Vec<_>>();
+        let parts = traceparent.split('-').collect::<Vec<_>>();
+        assert_eq!(parts[1], expected_parts[1]);
+        assert_ne!(parts[2], expected_parts[2]);
+        assert_eq!(trace.tracestate, expected_trace.tracestate);
     }
 
     async fn accept_websocket(listener: &TcpListener) -> WebSocketStream<TcpStream> {

codex-rs/exec-server/src/rpc.rs

@@ -342,6 +342,16 @@ impl RpcClient {
         drain_pending(&self.pending).await;
     }
 
+    #[tracing::instrument(
+        name = "codex.exec_server.request",
+        level = "info",
+        skip_all,
+        fields(
+            otel.kind = "client",
+            otel.name = method,
+            method,
+        )
+    )]
     pub(crate) async fn call<P, T>(&self, method: &str, params: &P) -> Result<T, RpcCallError>
     where
         P: Serialize,
@@ -803,6 +813,15 @@ mod tests {
             .expect("RPC response");
         assert_eq!(response, serde_json::json!({}));
         let trace = server.await.expect("server task").expect("trace context");
-        assert_eq!(trace, expected_trace);
+        let expected_traceparent = expected_trace
+            .traceparent
+            .as_deref()
+            .expect("parent traceparent");
+        let traceparent = trace.traceparent.as_deref().expect("request traceparent");
+        let expected_parts = expected_traceparent.split('-').collect::<Vec<_>>();
+        let parts = traceparent.split('-').collect::<Vec<_>>();
+        assert_eq!(parts[1], expected_parts[1]);
+        assert_ne!(parts[2], expected_parts[2]);
+        assert_eq!(trace.tracestate, expected_trace.tracestate);
     }
 }

codex-rs/ext/skills/src/state.rs

@@ -66,6 +66,12 @@ impl SkillsThreadState {
     /// Environment availability only controls whether the root is projected into the current
     /// step; it never invalidates the cache. There is intentionally no filesystem watcher or
     /// content-based invalidation because selected environment roots are treated as stable.
+    #[tracing::instrument(
+        name = "skills.executor.catalog_snapshot",
+        level = "info",
+        skip_all,
+        fields(root_count = query.executor_roots.len())
+    )]
     pub(crate) async fn executor_catalog_snapshot(
         &self,
         providers: &SkillProviders,
@@ -157,6 +163,7 @@ impl SkillsThreadState {
         next_cache
     }
 
+    #[tracing::instrument(name = "skills.executor.catalog_root", level = "info", skip_all)]
     async fn executor_root_catalog(
         &self,
         providers: &SkillProviders,