diff --git a/codex-rs/app-server/tests/common/json_logging.rs b/codex-rs/app-server/tests/common/json_logging.rs index c6c78bd0a476..8683d1bdaf1b 100644 --- a/codex-rs/app-server/tests/common/json_logging.rs +++ b/codex-rs/app-server/tests/common/json_logging.rs @@ -1,11 +1,79 @@ use std::path::Path; use std::process::Command; use std::process::Stdio; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; use anyhow::Context; use anyhow::Result; use serde_json::Value; use serde_json::json; +use tokio::sync::Notify; + +#[derive(Clone, Default)] +pub(crate) struct JsonLogCapture { + lines: Arc>>, + updated: Arc, +} + +impl JsonLogCapture { + pub(crate) fn record(&self, line: String) { + self.lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(line); + self.updated.notify_one(); + } + + pub(crate) async fn wait_for_event(&self, event_name: &str) -> Result { + let mut events = self.wait_for_events(event_name, /*count*/ 1).await?; + Ok(events.remove(0)) + } + + pub(crate) async fn wait_for_events( + &self, + event_name: &str, + count: usize, + ) -> Result> { + let result = tokio::time::timeout(Duration::from_secs(10), async { + loop { + let updated = self.updated.notified(); + let events = self + .events()? + .into_iter() + .filter(|event| event["fields"]["event.name"].as_str() == Some(event_name)) + .collect::>(); + if events.len() >= count { + return Ok(events); + } + updated.await; + } + }) + .await; + match result { + Ok(result) => result, + Err(_) => { + let lines = self + .lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .join("\n"); + anyhow::bail!( + "timed out waiting for {count} JSON log event(s) named `{event_name}`; captured stderr:\n{lines}" + ) + } + } + } + + pub(crate) fn events(&self) -> Result> { + let lines = self + .lines + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + json_log_events(lines.iter().map(String::as_str)) + } +} pub fn app_server_json_shutdown_event( binary: &str, @@ -31,25 +99,39 @@ pub fn app_server_json_shutdown_event( let stderr = String::from_utf8(output.stderr)?; anyhow::ensure!(output.status.success(), "app-server failed: {stderr}"); - let events = stderr - .lines() - .filter(|line| !line.is_empty()) - .map(serde_json::from_str::) - .collect::>>() - .with_context(|| format!("app-server stderr was not JSONL: {stderr}"))?; + let events = json_log_events(stderr.lines()) + .with_context(|| format!("app-server stderr was not valid JSONL: {stderr}"))?; let event = events .iter() .find(|event| event["fields"]["message"] == "processor task exited") .context("missing INFO shutdown event in app-server JSON logs")?; - let timestamp = event["timestamp"] - .as_str() - .context("shutdown event did not include a timestamp")?; - chrono::DateTime::parse_from_rfc3339(timestamp) - .with_context(|| format!("shutdown event timestamp was not RFC 3339: {timestamp}"))?; - Ok(json!({ "level": event["level"], "fields": event["fields"], "target": event["target"], })) } + +fn json_log_events<'a>(lines: impl IntoIterator) -> Result> { + lines + .into_iter() + .filter(|line| !line.is_empty()) + .map(|line| { + let event = serde_json::from_str::(line) + .with_context(|| format!("log line was not JSON: {line}"))?; + anyhow::ensure!( + event["level"].is_string() + && event["fields"].is_object() + && event["target"].is_string(), + "JSON log event did not include level, fields, and target: {line}" + ); + let timestamp = event["timestamp"] + .as_str() + .with_context(|| format!("JSON log event did not include a timestamp: {line}"))?; + chrono::DateTime::parse_from_rfc3339(timestamp).with_context(|| { + format!("JSON log event timestamp was not RFC 3339: {timestamp}") + })?; + Ok(event) + }) + .collect() +} diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index fbe43e7689b3..3f68dddf8126 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -123,6 +123,8 @@ use core_test_support::test_codex::TestEnv; use core_test_support::test_codex::test_env; use tokio::process::Command; +use crate::json_logging::JsonLogCapture; + pub struct TestAppServer { next_request_id: AtomicI64, /// Retain this child process until the client is dropped. The Tokio runtime @@ -134,6 +136,7 @@ pub struct TestAppServer { stdout: BufReader, pending_messages: VecDeque, auto_env: Option, + json_logs: JsonLogCapture, } pub const DEFAULT_CLIENT_NAME: &str = "codex-app-server-tests"; @@ -160,6 +163,14 @@ impl TestAppServer { /// URL-based configuration, this helper rejects a `codex_home` containing /// that file. pub async fn new_with_auto_env(codex_home: &Path) -> anyhow::Result { + Self::new_with_auto_env_and_env(codex_home, &[]).await + } + + /// Starts an auto-environment app server with child-process environment overrides. + pub async fn new_with_auto_env_and_env( + codex_home: &Path, + extra_env_overrides: &[(&str, Option<&str>)], + ) -> anyhow::Result { let environments_toml = codex_home.join("environments.toml"); ensure!( !environments_toml @@ -172,7 +183,7 @@ impl TestAppServer { let auto_env = test_env().await?; // Noise registry configuration takes precedence over the URL-based // provider, so clear inherited values to keep the selection hermetic. - let env_overrides = [ + let mut env_overrides = vec![ ( CODEX_EXEC_SERVER_URL_ENV_VAR, auto_env.environment().exec_server_url(), @@ -182,6 +193,7 @@ impl TestAppServer { (CODEX_EXEC_SERVER_NOISE_AUTH_TOKEN_ENV_VAR, None), (CODEX_EXEC_SERVER_NOISE_CHATGPT_ACCOUNT_ID_ENV_VAR, None), ]; + env_overrides.extend_from_slice(extra_env_overrides); let mut app_server = Self::new_with_env(codex_home, &env_overrides).await?; app_server.auto_env = Some(auto_env); Ok(app_server) @@ -202,6 +214,28 @@ impl TestAppServer { }) } + /// Waits for a JSON stderr event whose structured `event.name` field matches. + pub async fn wait_for_json_log_event( + &self, + event_name: &str, + ) -> anyhow::Result { + self.json_logs.wait_for_event(event_name).await + } + + /// Waits for the requested number of JSON stderr events with the same `event.name` field. + pub async fn wait_for_json_log_events( + &self, + event_name: &str, + count: usize, + ) -> anyhow::Result> { + self.json_logs.wait_for_events(event_name, count).await + } + + /// Returns every stderr line parsed and validated as a JSON log event. + pub fn json_log_events(&self) -> anyhow::Result> { + self.json_logs.events() + } + pub async fn new_without_managed_config(codex_home: &Path) -> anyhow::Result { Self::new_with_env(codex_home, &[(DISABLE_MANAGED_CONFIG_ENV_VAR, Some("1"))]).await } @@ -322,10 +356,13 @@ impl TestAppServer { // Forward child's stderr to our stderr so failures are visible even // when stdout/stderr are captured by the test harness. + let json_logs = JsonLogCapture::default(); if let Some(stderr) = process.stderr.take() { + let json_logs = json_logs.clone(); let mut stderr_reader = BufReader::new(stderr).lines(); tokio::spawn(async move { while let Ok(Some(line)) = stderr_reader.next_line().await { + json_logs.record(line.clone()); eprintln!("[mcp stderr] {line}"); } }); @@ -337,6 +374,7 @@ impl TestAppServer { stdout, pending_messages: VecDeque::new(), auto_env: None, + json_logs, }) } diff --git a/codex-rs/app-server/tests/suite/logging.rs b/codex-rs/app-server/tests/suite/logging.rs index ea2c31a76604..9967a56f9647 100644 --- a/codex-rs/app-server/tests/suite/logging.rs +++ b/codex-rs/app-server/tests/suite/logging.rs @@ -1,8 +1,28 @@ use anyhow::Result; +use app_test_support::TestAppServer; use app_test_support::app_server_json_shutdown_event; +use app_test_support::create_exec_command_sse_response; +use app_test_support::create_final_assistant_message_sse_response; +use app_test_support::create_mock_responses_server_sequence; +use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::TurnStartResponse; +use codex_app_server_protocol::UserInput; +use codex_features::Feature; +use core_test_support::skip_if_no_network; use pretty_assertions::assert_eq; use serde_json::json; +use std::collections::BTreeMap; use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +const READ_TIMEOUT: Duration = Duration::from_secs(10); #[test] fn standalone_app_server_emits_json_info_events() -> Result<()> { @@ -25,3 +45,174 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> { Ok(()) } + +#[tokio::test] +async fn app_server_emits_structured_tool_timing_events() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = create_mock_responses_server_sequence(vec![ + create_exec_command_sse_response("exec-call-1")?, + create_final_assistant_message_sse_response("done")?, + ]) + .await; + let codex_home = TempDir::new()?; + write_mock_responses_config_toml( + codex_home.path(), + &server.uri(), + &BTreeMap::from([(Feature::UnifiedExec, true)]), + /*auto_compact_limit*/ 100_000, + /*requires_openai_auth*/ None, + "mock_provider", + "compact", + )?; + + let mut app_server = TestAppServer::new_with_auto_env_and_env( + codex_home.path(), + &[ + ("LOG_FORMAT", Some("json")), + ( + "RUST_LOG", + Some( + "warn,codex_core::tools::parallel=info,codex_core::turn_timing=info,codex_otel.log_only=info", + ), + ), + ], + ) + .await?; + timeout(READ_TIMEOUT, app_server.initialize()).await??; + + let thread_start_id = app_server + .send_thread_start_request_with_auto_env(ThreadStartParams { + model: Some("mock-model".to_string()), + ..Default::default() + }) + .await?; + let thread_start_response: JSONRPCResponse = timeout( + READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(thread_start_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response(thread_start_response)?; + + let turn_start_id = app_server + .send_turn_start_request(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "run a command".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }) + .await?; + let turn_start_response: JSONRPCResponse = timeout( + READ_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(turn_start_id)), + ) + .await??; + let TurnStartResponse { turn } = to_response(turn_start_response)?; + + timeout( + READ_TIMEOUT, + app_server.read_stream_until_notification_message("turn/completed"), + ) + .await??; + + let tool_call = app_server + .wait_for_json_log_event("codex.tool_call") + .await?; + assert_eq!(tool_call["level"], "INFO"); + assert_eq!(tool_call["target"], "codex_core::tools::parallel"); + assert_eq!(tool_call["fields"]["message"], "tool call completed"); + assert!(tool_call["fields"]["trace_id"].is_string()); + assert_eq!(tool_call["fields"]["conversation.id"], thread.id); + assert_eq!(tool_call["fields"]["turn_id"], turn.id); + assert_eq!(tool_call["fields"]["tool_name"], "exec_command"); + assert_eq!(tool_call["fields"]["call_id"], "exec-call-1"); + assert_eq!(tool_call["fields"]["tool_source"], "direct"); + assert_eq!(tool_call["fields"]["code_mode_cell_id"], ""); + assert_eq!(tool_call["fields"]["code_mode_runtime_tool_call_id"], ""); + assert_eq!(tool_call["fields"]["execution_started"], true); + assert_nonnegative_duration_fields( + &tool_call, + &[ + "dispatch_duration_ms", + "handler_duration_ms", + "total_duration_ms", + ], + ); + let dispatch_duration = duration_field(&tool_call, "dispatch_duration_ms"); + let handler_duration = duration_field(&tool_call, "handler_duration_ms"); + let total_duration = duration_field(&tool_call, "total_duration_ms"); + assert!(total_duration > 0.0); + let truncation_delta = total_duration - dispatch_duration - handler_duration; + assert!( + (0.0..=1.0).contains(&truncation_delta), + "dispatch and handler durations must account for total duration within integer truncation: {tool_call}" + ); + + let tool_result = app_server + .wait_for_json_log_event("codex.tool_result") + .await?; + assert_eq!(tool_result["level"], "INFO"); + assert_eq!(tool_result["target"], "codex_otel.log_only"); + assert!(tool_result["fields"]["trace_id"].is_string()); + assert_eq!(tool_result["fields"]["conversation.id"], thread.id); + assert_eq!(tool_result["fields"]["turn_id"], turn.id); + assert_eq!(tool_result["fields"]["tool_name"], "exec_command"); + assert_eq!(tool_result["fields"]["call_id"], "exec-call-1"); + assert_eq!(tool_result["fields"]["tool_source"], "direct"); + assert!( + matches!( + tool_result["fields"]["success"].as_str(), + Some("true" | "false") + ), + "success must be a boolean string: {tool_result}" + ); + assert!( + tool_result["fields"]["duration_ms"] + .as_str() + .is_some_and(|duration_ms| duration_ms.parse::().is_ok()), + "duration_ms must retain its pre-existing integer-string schema: {tool_result}" + ); + + let inferences = app_server + .wait_for_json_log_events("codex.inference", /*count*/ 2) + .await?; + assert_eq!( + inferences + .iter() + .map(|event| event["fields"]["inference_index"].as_u64()) + .collect::>(), + vec![Some(1), Some(2)] + ); + for inference in inferences { + assert_eq!(inference["level"], "INFO"); + assert_eq!(inference["target"], "codex_core::turn_timing"); + assert_eq!(inference["fields"]["message"], "inference completed"); + assert!(inference["fields"]["trace_id"].is_string()); + assert_eq!(inference["fields"]["conversation.id"], thread.id); + assert_eq!(inference["fields"]["turn_id"], turn.id); + assert_eq!(inference["fields"]["model"], "mock-model"); + assert_eq!( + inference["fields"]["provider_name"], + "Mock provider for test" + ); + assert_eq!(inference["fields"]["result"], "success"); + assert_nonnegative_duration_fields(&inference, &["duration_ms"]); + } + + Ok(()) +} + +fn assert_nonnegative_duration_fields(event: &serde_json::Value, fields: &[&str]) { + for field in fields { + let duration = duration_field(event, field); + assert!(duration >= 0.0, "{field} must be nonnegative: {event}"); + } +} + +fn duration_field(event: &serde_json::Value, field: &str) -> f64 { + event["fields"][field] + .as_f64() + .unwrap_or_else(|| panic!("{field} must be a JSON number: {event}")) +} diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index f16b525002fc..c8880cb462a6 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -65,6 +65,7 @@ use crate::tools::router::extension_tool_executors; use crate::tools::spec_plan::search_tool_enabled; use crate::tools::spec_plan::tool_suggest_enabled; use crate::turn_diff_tracker::TurnDiffTracker; +use crate::turn_timing::InferenceTimingResult; use crate::turn_timing::record_turn_ttft_metric; use crate::util::error_or_panic; use codex_analytics::AppInvocation; @@ -1903,13 +1904,19 @@ async fn try_run_sampling_request( auth_mode = sess.services.auth_manager.auth_mode(), features = sess.features.enabled_features(), ); + let provider_info = turn_context.provider.info(); let inference_trace = sess.services.rollout_thread_trace.inference_trace_context( turn_context.sub_id.as_str(), turn_context.model_info.slug.as_str(), - turn_context.provider.info().name.as_str(), + provider_info.name.as_str(), ); - let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling(); - let mut stream = client_session + let sampling_timing_guard = turn_context.turn_timing_state.begin_sampling( + &sess.thread_id, + &turn_context.sub_id, + &turn_context.model_info.slug, + &provider_info.name, + ); + let stream_result = client_session .stream( prompt, &turn_context.model_info, @@ -1922,7 +1929,18 @@ async fn try_run_sampling_request( ) .instrument(trace_span!("stream_request")) .or_cancel(&cancellation_token) - .await??; + .await; + let mut stream = match stream_result { + Ok(Ok(stream)) => stream, + Ok(Err(err)) => { + sampling_timing_guard.finish_inference(InferenceTimingResult::Error); + return Err(err); + } + Err(codex_async_utils::CancelErr::Cancelled) => { + sampling_timing_guard.finish_inference(InferenceTimingResult::Cancelled); + return Err(CodexErr::TurnAborted); + } + }; let mut in_flight: FuturesOrdered>> = FuturesOrdered::new(); let mut needs_follow_up = false; @@ -2346,7 +2364,11 @@ async fn try_run_sampling_request( } } }; - drop(sampling_timing_guard); + sampling_timing_guard.finish_inference(match &outcome { + Ok(_) => InferenceTimingResult::Success, + Err(CodexErr::TurnAborted) => InferenceTimingResult::Cancelled, + Err(_) => InferenceTimingResult::Error, + }); flush_assistant_text_segments_all( &sess, diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index a2ee89ab837d..e45277d3c739 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::OnceLock; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Instant; @@ -9,6 +10,7 @@ use tokio_util::either::Either; use tokio_util::sync::CancellationToken; use tokio_util::task::AbortOnDropHandle; use tracing::Instrument; +use tracing::info; use tracing::instrument; use tracing::trace_span; @@ -27,6 +29,16 @@ use crate::tools::router::ToolRouter; use codex_protocol::error::CodexErr; use codex_protocol::models::ResponseInputItem; +struct ToolCallTimingGuard { + started_at: Instant, + execution_started_at: Arc>, + conversation_id: String, + turn_id: String, + call_id: String, + tool_name: codex_tools::ToolName, + source: ToolCallSource, +} + #[derive(Clone)] pub(crate) struct ToolCallRuntime { router: Arc, @@ -96,6 +108,11 @@ impl ToolCallRuntime { let invocation_cancellation_token = cancellation_token.clone(); let wait_for_runtime_cancellation = self.router.tool_waits_for_runtime_cancellation(&call); let started = Instant::now(); + let tool_call_timing_guard = + ToolCallTimingGuard::capture(started, &session.thread_id, &turn.sub_id, &call, &source); + let execution_started_at = tool_call_timing_guard + .as_ref() + .map(|timing| Arc::clone(&timing.execution_started_at)); let abort_session = Arc::clone(&session); let abort_source = source.clone(); let abort_turn = Arc::clone(&turn); @@ -119,6 +136,9 @@ impl ToolCallRuntime { } else { Either::Right(lock.write().await) }; + if let Some(execution_started_at) = execution_started_at { + let _ = execution_started_at.set(Instant::now()); + } router .dispatch_tool_call_with_terminal_outcome( @@ -135,6 +155,7 @@ impl ToolCallRuntime { })); async move { + let _tool_call_timing_guard = tool_call_timing_guard; tokio::select! { res = &mut handle => res.map_err(Self::tool_task_join_error)?, _ = cancellation_token.cancelled() => { @@ -237,6 +258,74 @@ impl ToolCallRuntime { } } +impl ToolCallTimingGuard { + fn capture( + started_at: Instant, + conversation_id: &impl std::fmt::Display, + turn_id: &str, + call: &ToolCall, + source: &ToolCallSource, + ) -> Option { + if !tracing::enabled!(tracing::Level::INFO) { + return None; + } + + Some(Self { + started_at, + execution_started_at: Arc::new(OnceLock::new()), + conversation_id: conversation_id.to_string(), + turn_id: turn_id.to_string(), + call_id: call.call_id.clone(), + tool_name: call.tool_name.clone(), + source: source.clone(), + }) + } +} + +impl Drop for ToolCallTimingGuard { + fn drop(&mut self) { + let completed_at = Instant::now(); + info!( + event.name = "codex.tool_call", + trace_id = %codex_otel::current_span_trace_id().unwrap_or_default(), + conversation.id = %self.conversation_id, + turn_id = %self.turn_id, + tool_name = %self.tool_name, + call_id = %self.call_id, + tool_source = match &self.source { + ToolCallSource::Direct => "direct", + ToolCallSource::CodeMode { .. } => "code_mode", + }, + code_mode_cell_id = match &self.source { + ToolCallSource::Direct => "", + ToolCallSource::CodeMode { cell_id, .. } => cell_id.as_str(), + }, + code_mode_runtime_tool_call_id = match &self.source { + ToolCallSource::Direct => "", + ToolCallSource::CodeMode { runtime_tool_call_id, .. } => runtime_tool_call_id.as_str(), + }, + execution_started = self.execution_started_at.get().is_some(), + dispatch_duration_ms = self.execution_started_at.get().map_or_else( + || u64::try_from(completed_at.duration_since(self.started_at).as_millis()).unwrap_or(u64::MAX), + |execution_started_at| { + u64::try_from(execution_started_at.duration_since(self.started_at).as_millis()) + .unwrap_or(u64::MAX) + }, + ), + handler_duration_ms = self.execution_started_at.get().map_or( + 0, + |execution_started_at| { + u64::try_from(completed_at.duration_since(*execution_started_at).as_millis()) + .unwrap_or(u64::MAX) + }, + ), + total_duration_ms = u64::try_from(completed_at.duration_since(self.started_at).as_millis()) + .unwrap_or(u64::MAX), + "tool call completed" + ); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index 2220d4244a9c..5895208d0e07 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -14,6 +14,7 @@ use crate::sandbox_tags::permission_profile_policy_tag; use crate::sandbox_tags::permission_profile_sandbox_tag; use crate::session::turn_context::TurnContext; use crate::tools::context::FunctionToolOutput; +use crate::tools::context::ToolCallSource; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; @@ -444,6 +445,8 @@ impl ToolRegistry { None => { let message = unsupported_tool_call_message(&invocation.payload, &tool_name); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = + tool_result_trace_fields(&invocation.turn.sub_id, &invocation.source, "", ""); otel.tool_result_with_tags( tool_name_flat.as_ref(), &call_id_owned, @@ -452,7 +455,7 @@ impl ToolRegistry { /*success*/ false, &message, &base_tool_result_tags, - /*extra_trace_fields*/ &[], + &extra_trace_fields, ); let err = FunctionCallError::RespondToModel(message); dispatch_trace.record_failed(&err); @@ -463,18 +466,25 @@ impl ToolRegistry { let telemetry_tags = tool.telemetry_tags(&invocation).await; let mut tool_result_tags = Vec::with_capacity(base_tool_result_tags.len() + telemetry_tags.len()); - let mut extra_trace_fields = Vec::new(); + let mut mcp_server = ""; + let mut mcp_server_origin = ""; tool_result_tags.extend_from_slice(&base_tool_result_tags); for (key, value) in &telemetry_tags { - if matches!(*key, "mcp_server" | "mcp_server_origin") { - extra_trace_fields.push((*key, value.as_str())); - } else { - tool_result_tags.push((*key, value.as_str())); + match *key { + "mcp_server" => mcp_server = value.as_str(), + "mcp_server_origin" => mcp_server_origin = value.as_str(), + _ => tool_result_tags.push((*key, value.as_str())), } } if !tool.matches_kind(&invocation.payload) { let message = format!("tool {tool_name} invoked with incompatible payload"); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = tool_result_trace_fields( + &invocation.turn.sub_id, + &invocation.source, + mcp_server, + mcp_server_origin, + ); otel.tool_result_with_tags( tool_name_flat.as_ref(), &call_id_owned, @@ -541,6 +551,12 @@ impl ToolRegistry { let response_cell = tokio::sync::Mutex::new(None); let invocation_for_tool = invocation.clone(); let log_payload = invocation.payload.log_payload(); + let extra_trace_fields = tool_result_trace_fields( + &invocation.turn.sub_id, + &invocation.source, + mcp_server, + mcp_server_origin, + ); let result = otel .log_tool_result_with_tags( @@ -670,6 +686,32 @@ impl ToolRegistry { } } +fn tool_result_trace_fields<'a>( + turn_id: &'a str, + source: &'a ToolCallSource, + mcp_server: &'a str, + mcp_server_origin: &'a str, +) -> [(&'static str, &'a str); 6] { + let (tool_source, code_mode_cell_id, code_mode_runtime_tool_call_id) = match source { + ToolCallSource::Direct => ("direct", "", ""), + ToolCallSource::CodeMode { + cell_id, + runtime_tool_call_id, + } => ("code_mode", cell_id.as_str(), runtime_tool_call_id.as_str()), + }; + [ + ("turn_id", turn_id), + ("tool_source", tool_source), + ("code_mode_cell_id", code_mode_cell_id), + ( + "code_mode_runtime_tool_call_id", + code_mode_runtime_tool_call_id, + ), + ("mcp_server", mcp_server), + ("mcp_server_origin", mcp_server_origin), + ] +} + async fn notify_tool_finish_if_unclaimed( invocation: &ToolInvocation, terminal_outcome_reached: Option<&AtomicBool>, diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 04f7aeb69be9..1b82598f7a52 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -10,6 +10,7 @@ use codex_otel::TURN_TTFM_DURATION_METRIC; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use tokio::sync::Mutex; +use tracing::info; use crate::ResponseEvent; use crate::session::turn_context::TurnContext; @@ -80,6 +81,25 @@ pub(crate) struct TurnProfileTimingGuard { timing: Arc, phase: TurnProfilePhase, active: bool, + inference_log_context: Option, +} + +#[derive(Clone, Copy)] +pub(crate) enum InferenceTimingResult { + Success, + Cancelled, + Error, +} + +struct InferenceLogContext { + started_at: Instant, + conversation_id: String, + turn_id: String, + model: String, + provider_name: String, + trace_id: String, + inference_index: u32, + result: InferenceTimingResult, } impl TurnTimingState { @@ -118,12 +138,35 @@ impl TurnTimingState { self.profile_state().complete(Instant::now()) } - pub(crate) fn begin_sampling(self: &Arc) -> TurnProfileTimingGuard { - let active = self.profile_state().begin_sampling(Instant::now()); + pub(crate) fn begin_sampling( + self: &Arc, + conversation_id: &impl std::fmt::Display, + turn_id: &str, + model: &str, + provider_name: &str, + ) -> TurnProfileTimingGuard { + let started_at = Instant::now(); + let inference_index = self.profile_state().begin_sampling(started_at); + let active = inference_index.is_some(); + let inference_log_context = if tracing::enabled!(tracing::Level::INFO) { + inference_index.map(|inference_index| InferenceLogContext { + started_at, + conversation_id: conversation_id.to_string(), + turn_id: turn_id.to_string(), + model: model.to_string(), + provider_name: provider_name.to_string(), + trace_id: codex_otel::current_span_trace_id().unwrap_or_default(), + inference_index, + result: InferenceTimingResult::Error, + }) + } else { + None + }; TurnProfileTimingGuard { timing: Arc::clone(self), phase: TurnProfilePhase::Sampling, active, + inference_log_context, } } @@ -137,6 +180,7 @@ impl TurnTimingState { timing: Arc::clone(self), phase: TurnProfilePhase::ToolBlocking, active, + inference_log_context: None, } } @@ -166,6 +210,14 @@ impl TurnTimingState { } } +impl TurnProfileTimingGuard { + pub(crate) fn finish_inference(mut self, result: InferenceTimingResult) { + if let Some(log_context) = &mut self.inference_log_context { + log_context.result = result; + } + } +} + impl Drop for TurnProfileTimingGuard { fn drop(&mut self) { if self.active { @@ -173,6 +225,25 @@ impl Drop for TurnProfileTimingGuard { .profile_state() .end_phase(Instant::now(), self.phase); } + if let Some(log_context) = &self.inference_log_context { + info!( + event.name = "codex.inference", + trace_id = %log_context.trace_id, + conversation.id = %log_context.conversation_id, + turn_id = %log_context.turn_id, + model = %log_context.model, + provider_name = %log_context.provider_name, + inference_index = log_context.inference_index, + result = match log_context.result { + InferenceTimingResult::Success => "success", + InferenceTimingResult::Cancelled => "cancelled", + InferenceTimingResult::Error => "error", + }, + duration_ms = u64::try_from(log_context.started_at.elapsed().as_millis()) + .unwrap_or(u64::MAX), + "inference completed" + ); + } } } @@ -200,12 +271,12 @@ impl TurnProfileState { }; } - fn begin_sampling(&mut self, now: Instant) -> bool { + fn begin_sampling(&mut self, now: Instant) -> Option { if self.completed_profile.is_some() || self.started_at.is_none() || self.active_phase.is_some() { - return false; + return None; } self.advance(now); if self.seen_sampling { @@ -214,7 +285,7 @@ impl TurnProfileState { self.seen_sampling = true; self.active_phase = Some(TurnProfilePhase::Sampling); self.sampling_request_count = self.sampling_request_count.saturating_add(1); - true + Some(self.sampling_request_count) } fn record_sampling_retry(&mut self) { diff --git a/codex-rs/core/tests/suite/otel.rs b/codex-rs/core/tests/suite/otel.rs index e3415d945ba2..97c7f6b5cb2b 100644 --- a/codex-rs/core/tests/suite/otel.rs +++ b/codex-rs/core/tests/suite/otel.rs @@ -710,7 +710,7 @@ async fn turn_and_completed_response_spans_record_token_usage() { ); } -#[tokio::test] +#[tokio::test(flavor = "current_thread")] async fn handle_responses_span_records_response_kind_and_tool_name() { let buffer: &'static Mutex> = Box::leak(Box::new(Mutex::new(Vec::new()))); let subscriber = tracing_subscriber::fmt() diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index 3c488f0ad7c3..08d2d0319ff1 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -334,7 +334,7 @@ impl LocalProcess { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(self.inner.telemetry.process_started()), + metrics: Some(self.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: prepared.sandbox, sandbox_denied: false, @@ -1306,7 +1306,7 @@ mod tests { output_notify: Arc::clone(&output_notify), open_streams: 2, closed: false, - metrics: Some(backend.inner.telemetry.process_started()), + metrics: Some(backend.inner.telemetry.process_started(process_id.as_ref())), termination_requested: false, sandbox: SandboxType::None, sandbox_denied: false, diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 63f6bd45d9a6..56690c65c510 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -128,6 +128,13 @@ async fn run_connection( JsonRpcConnectionEvent::Message(message) => match message { codex_exec_server_protocol::JSONRPCMessage::Request(request) => { let request_started_at = Instant::now(); + let request_log_context = telemetry.request_log_context( + &request.id, + request + .trace + .as_ref() + .and_then(|trace| trace.traceparent.as_deref()), + ); if let Some((method, route)) = router.request_route(request.method.as_str()) { let request_span = request_span(method, &request); let message = tokio::select! { @@ -135,6 +142,7 @@ async fn run_connection( _ = disconnected_rx.changed() => { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -149,6 +157,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -156,7 +165,12 @@ async fn run_connection( break; } request_span.record("result", result); - telemetry.request_completed(method, result, request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + result, + request_started_at.elapsed(), + ); drop(request_span); } else { let method = "unknown"; @@ -174,6 +188,7 @@ async fn run_connection( { request_span.record("result", "disconnected"); telemetry.request_completed( + request_log_context.as_ref(), method, "disconnected", request_started_at.elapsed(), @@ -181,7 +196,12 @@ async fn run_connection( break; } request_span.record("result", "error"); - telemetry.request_completed(method, "error", request_started_at.elapsed()); + telemetry.request_completed( + request_log_context.as_ref(), + method, + "error", + request_started_at.elapsed(), + ); } } codex_exec_server_protocol::JSONRPCMessage::Notification(notification) => { diff --git a/codex-rs/exec-server/src/telemetry.rs b/codex-rs/exec-server/src/telemetry.rs index 650faacb2c63..6191b55bf0ae 100644 --- a/codex-rs/exec-server/src/telemetry.rs +++ b/codex-rs/exec-server/src/telemetry.rs @@ -4,6 +4,7 @@ use std::time::Duration; use std::time::Instant; use codex_otel::MetricsClient; +use tracing::info; use tracing::warn; const CONNECTIONS_ACTIVE_METRIC: &str = "exec_server_connections_active"; @@ -96,10 +97,21 @@ pub(crate) struct ConnectionMetricGuard { pub(crate) struct ProcessMetricGuard { telemetry: ExecServerTelemetry, + log_context: Option, started_at: Instant, result: &'static str, } +struct ProcessLogContext { + process_id: String, + trace_id: String, +} + +pub(crate) struct RequestLogContext { + request_id: String, + traceparent: Option, +} + impl ExecServerTelemetry { pub fn new(metrics: MetricsClient) -> Self { let active = Arc::new(Mutex::new(ActiveCounts::default())); @@ -129,10 +141,22 @@ impl ExecServerTelemetry { pub(crate) fn request_completed( &self, + log_context: Option<&RequestLogContext>, method: &'static str, result: &'static str, duration: Duration, ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_request", + request_id = %log_context.request_id, + method, + result, + duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX), + traceparent = log_context.traceparent.as_deref().unwrap_or(""), + "exec-server request completed" + ); + } self.with_inner(|inner| { let tags = [("method", method), ("result", result)]; inner.counter(REQUESTS_TOTAL_METRIC, REQUESTS_TOTAL_DESCRIPTION, &tags); @@ -145,6 +169,17 @@ impl ExecServerTelemetry { }); } + pub(crate) fn request_log_context( + &self, + request_id: &impl std::fmt::Display, + traceparent: Option<&str>, + ) -> Option { + self.info_events_enabled().then(|| RequestLogContext { + request_id: request_id.to_string(), + traceparent: traceparent.map(str::to_string), + }) + } + pub(crate) fn remote_registration_completed(&self, result: &'static str, duration: Duration) { self.record_operation(REMOTE_REGISTRATION_METRICS, result, duration); } @@ -163,18 +198,37 @@ impl ExecServerTelemetry { }); } - pub(crate) fn process_started(&self) -> ProcessMetricGuard { + pub(crate) fn process_started(&self, process_id: &str) -> ProcessMetricGuard { self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ 1); }); ProcessMetricGuard { telemetry: self.clone(), + log_context: self.info_events_enabled().then(|| ProcessLogContext { + process_id: process_id.to_string(), + trace_id: codex_otel::current_span_trace_id().unwrap_or_default(), + }), started_at: Instant::now(), result: "unknown", } } - fn process_finished(&self, result: &'static str, duration: Duration) { + fn process_finished( + &self, + log_context: Option<&ProcessLogContext>, + result: &'static str, + duration: Duration, + ) { + if let Some(log_context) = log_context { + info!( + event.name = "codex.exec_server_process", + process_id = %log_context.process_id, + trace_id = %log_context.trace_id, + result, + duration_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX), + "exec-server process completed" + ); + } self.with_inner(|inner| { inner.adjust_process_count(/*delta*/ -1); inner.counter( @@ -191,6 +245,10 @@ impl ExecServerTelemetry { }); } + pub(crate) fn info_events_enabled(&self) -> bool { + tracing::enabled!(tracing::Level::INFO) + } + fn connection_finished(&self, transport: ConnectionTransport) { self.with_inner(|inner| { inner.adjust_connection_count(transport, /*delta*/ -1); @@ -236,8 +294,11 @@ impl ProcessMetricGuard { impl Drop for ProcessMetricGuard { fn drop(&mut self) { - self.telemetry - .process_finished(self.result, self.started_at.elapsed()); + self.telemetry.process_finished( + self.log_context.as_ref(), + self.result, + self.started_at.elapsed(), + ); } } @@ -338,3 +399,7 @@ fn register_active_gauge( warn!(metric = name, "failed to register exec-server gauge"); } } + +#[cfg(test)] +#[path = "telemetry_tests.rs"] +mod tests; diff --git a/codex-rs/exec-server/src/telemetry_tests.rs b/codex-rs/exec-server/src/telemetry_tests.rs new file mode 100644 index 000000000000..3caff2a93b9d --- /dev/null +++ b/codex-rs/exec-server/src/telemetry_tests.rs @@ -0,0 +1,170 @@ +use std::collections::BTreeMap; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; + +use pretty_assertions::assert_eq; +use tracing::Event; +use tracing::Level; +use tracing::Subscriber; +use tracing::field::Field; +use tracing::field::Visit; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::prelude::*; +use tracing_subscriber::registry::LookupSpan; + +use super::ExecServerTelemetry; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct CapturedEvent { + level: Level, + target: String, + fields: BTreeMap, +} + +#[derive(Clone, Default)] +struct CaptureLayer { + events: Arc>>, +} + +impl CaptureLayer { + fn events(&self) -> Vec { + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +impl Layer for CaptureLayer +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) { + let mut visitor = FieldVisitor::default(); + event.record(&mut visitor); + self.events + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(CapturedEvent { + level: *event.metadata().level(), + target: event.metadata().target().to_string(), + fields: visitor.fields, + }); + } +} + +#[derive(Default)] +struct FieldVisitor { + fields: BTreeMap, +} + +impl Visit for FieldVisitor { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.fields + .insert(field.name().to_string(), format!("{value:?}")); + } + + fn record_str(&mut self, field: &Field, value: &str) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } + + fn record_f64(&mut self, field: &Field, value: f64) { + self.fields + .insert(field.name().to_string(), value.to_string()); + } +} + +#[test] +fn exec_server_timing_events_are_structured_info_logs() { + let capture = CaptureLayer::default(); + let subscriber = tracing_subscriber::registry().with(capture.clone()); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + let request_log_context = telemetry + .request_log_context(&"request-1", Some("00-trace-parent")) + .expect("INFO event should be enabled"); + telemetry.request_completed( + Some(&request_log_context), + "process/start", + "success", + Duration::from_millis(42), + ); + telemetry.process_started("process-1").finish("success"); + }); + + let events = capture.events(); + let request = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_request") + }) + .expect("request timing event"); + assert_eq!(request.level, Level::INFO); + assert_eq!(request.target, "codex_exec_server::telemetry"); + assert_eq!( + request.fields, + BTreeMap::from([ + ("duration_ms".to_string(), "42".to_string()), + ( + "event.name".to_string(), + "codex.exec_server_request".to_string(), + ), + ( + "message".to_string(), + "exec-server request completed".to_string(), + ), + ("method".to_string(), "process/start".to_string()), + ("request_id".to_string(), "request-1".to_string()), + ("result".to_string(), "success".to_string()), + ("traceparent".to_string(), "00-trace-parent".to_string()), + ]) + ); + + let process = events + .iter() + .find(|event| { + event + .fields + .get("event.name") + .is_some_and(|name| name == "codex.exec_server_process") + }) + .expect("process timing event"); + assert_eq!(process.level, Level::INFO); + assert_eq!(process.target, "codex_exec_server::telemetry"); + assert_eq!( + process.fields.get("message").map(String::as_str), + Some("exec-server process completed") + ); + assert_eq!( + process.fields.get("process_id").map(String::as_str), + Some("process-1") + ); + assert_eq!(process.fields.get("trace_id").map(String::as_str), Some("")); + assert_eq!( + process.fields.get("result").map(String::as_str), + Some("success") + ); + assert!(process.fields.contains_key("duration_ms")); +} + +#[test] +fn disabled_info_events_do_not_capture_process_log_values() { + let subscriber = tracing_subscriber::registry() + .with(tracing_subscriber::filter::filter_fn(|_metadata| false)); + + tracing::subscriber::with_default(subscriber, || { + tracing::callsite::rebuild_interest_cache(); + let telemetry = ExecServerTelemetry::default(); + assert!(!telemetry.info_events_enabled()); + let process = telemetry.process_started("process-1"); + assert!(process.log_context.is_none()); + }); +} diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 8210c0e5b984..d09633165f64 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -1062,7 +1062,14 @@ impl SessionTelemetry { log_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = "", tool_name = %tool_name, + call_id = "", + tool_source = "", + code_mode_cell_id = "", + code_mode_runtime_tool_call_id = "", + arguments = "", duration_ms = %Duration::ZERO.as_millis(), success = %false, output = %error, @@ -1072,7 +1079,13 @@ impl SessionTelemetry { trace_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = "", tool_name = %tool_name, + call_id = "", + tool_source = "", + code_mode_cell_id = "", + code_mode_runtime_tool_call_id = "", duration_ms = %Duration::ZERO.as_millis(), success = %false, output_length = error.len() as i64, @@ -1101,33 +1114,40 @@ impl SessionTelemetry { tags.extend_from_slice(extra_tags); self.counter(TOOL_CALL_COUNT_METRIC, /*inc*/ 1, &tags); self.record_duration(TOOL_CALL_DURATION_METRIC, duration, &tags); - let mcp_server = trace_field_value(extra_trace_fields, "mcp_server").unwrap_or(""); - let mcp_server_origin = - trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or(""); log_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""), tool_name = %tool_name, call_id = %call_id, + tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""), + code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""), + code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""), arguments = %arguments, duration_ms = %duration.as_millis(), success = %success_str, output = %output, - mcp_server = %mcp_server, - mcp_server_origin = %mcp_server_origin, + mcp_server = %trace_field_value(extra_trace_fields, "mcp_server").unwrap_or(""), + mcp_server_origin = %trace_field_value(extra_trace_fields, "mcp_server_origin").unwrap_or(""), ); trace_event!( self, event.name = "codex.tool_result", + trace_id = %crate::current_span_trace_id().unwrap_or_default(), + turn_id = %trace_field_value(extra_trace_fields, "turn_id").unwrap_or(""), tool_name = %tool_name, call_id = %call_id, + tool_source = %trace_field_value(extra_trace_fields, "tool_source").unwrap_or(""), + code_mode_cell_id = %trace_field_value(extra_trace_fields, "code_mode_cell_id").unwrap_or(""), + code_mode_runtime_tool_call_id = %trace_field_value(extra_trace_fields, "code_mode_runtime_tool_call_id").unwrap_or(""), duration_ms = %duration.as_millis(), success = %success_str, arguments_length = arguments.len() as i64, output_length = output.len() as i64, output_line_count = output.lines().count() as i64, - tool_origin = if mcp_server.is_empty() { "builtin" } else { "mcp" }, - mcp_tool = !mcp_server.is_empty(), + tool_origin = if trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty() { "builtin" } else { "mcp" }, + mcp_tool = !trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("").is_empty(), ); } diff --git a/codex-rs/otel/tests/suite/otel_export_routing_policy.rs b/codex-rs/otel/tests/suite/otel_export_routing_policy.rs index 17b6f764dbd3..ffff814fc5e9 100644 --- a/codex-rs/otel/tests/suite/otel_export_routing_policy.rs +++ b/codex-rs/otel/tests/suite/otel_export_routing_policy.rs @@ -255,6 +255,10 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() { &[ ("mcp_server", "internal-mcp"), ("mcp_server_origin", "stdio"), + ("turn_id", "turn-1"), + ("tool_source", "code_mode"), + ("code_mode_cell_id", "cell-1"), + ("code_mode_runtime_tool_call_id", "runtime-call-1"), ], ); }); @@ -286,6 +290,29 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() { tool_log_attrs.get("mcp_server_origin").map(String::as_str), Some("stdio") ); + assert_eq!( + tool_log_attrs.get("turn_id").map(String::as_str), + Some("turn-1") + ); + assert_eq!( + tool_log_attrs.get("tool_source").map(String::as_str), + Some("code_mode") + ); + assert_eq!( + tool_log_attrs.get("code_mode_cell_id").map(String::as_str), + Some("cell-1") + ); + assert_eq!( + tool_log_attrs + .get("code_mode_runtime_tool_call_id") + .map(String::as_str), + Some("runtime-call-1") + ); + assert_eq!( + tool_log_attrs.get("duration_ms").map(String::as_str), + Some("42") + ); + assert!(tool_log_attrs.contains_key("trace_id")); let spans = span_exporter.get_finished_spans().expect("span export"); assert_eq!(spans.len(), 1);