diff --git a/codex-rs/core/src/session/extension_interruption.rs b/codex-rs/core/src/session/extension_interruption.rs index 87da41f206..70f2d46df2 100644 --- a/codex-rs/core/src/session/extension_interruption.rs +++ b/codex-rs/core/src/session/extension_interruption.rs @@ -13,6 +13,7 @@ use codex_protocol::protocol::TurnAbortReason; use std::sync::Arc; use tokio::sync::oneshot; use tokio::sync::oneshot::error::TryRecvError; +use tracing::Instrument; #[cfg(test)] #[path = "extension_interruption_tests.rs"] @@ -108,16 +109,21 @@ impl Session { let runtime = self.services.runtime_handle.clone(); let session = Arc::clone(self); let turn_id = turn_id.to_owned(); - drop(runtime.spawn(async move { - if session - .abort_turn_if_active(&turn_id, TurnAbortReason::Interrupted, error) - .await - { - // Extension aborts bypass normal task completion; user interrupts do not. - session - .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Interrupted) - .await; - } - })); + drop( + runtime.spawn( + async move { + if session + .abort_turn_if_active(&turn_id, TurnAbortReason::Interrupted, error) + .await + { + // Extension aborts bypass normal task completion; user interrupts do not. + session + .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Interrupted) + .await; + } + } + .in_current_span(), + ), + ); } } diff --git a/codex-rs/core/src/session/extension_interruption_tests.rs b/codex-rs/core/src/session/extension_interruption_tests.rs index 7e46b02e35..39e038bc7a 100644 --- a/codex-rs/core/src/session/extension_interruption_tests.rs +++ b/codex-rs/core/src/session/extension_interruption_tests.rs @@ -1,4 +1,5 @@ -//! Regression coverage for cancellation while submitting a conditional interrupt. +//! Regression coverage for conditional interruption and detached abort tracing. +#![cfg(test)] use super::*; use crate::session::SessionIo; @@ -12,10 +13,104 @@ use codex_protocol::protocol::AgentStatus; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; +use opentelemetry::trace::TracerProvider; +use opentelemetry_sdk::trace::InMemorySpanExporter; +use opentelemetry_sdk::trace::SdkTracerProvider; use pretty_assertions::assert_eq; use std::time::Duration; use tokio::sync::Notify; use tokio::sync::watch; +use tracing::Instrument; +use tracing_subscriber::prelude::*; + +#[tokio::test] +async fn interrupt_with_warning_exports_turn_profile_under_caller_span() -> anyhow::Result<()> { + let exported = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exported.clone()) + .build(); + let subscriber = tracing_subscriber::registry().with( + tracing_opentelemetry::layer() + .with_tracer(provider.tracer("guardian-interrupt-test")) + .with_filter(tracing_subscriber::filter::filter_fn( + codex_otel::OtelProvider::trace_export_filter, + )), + ); + let _interest_cache_guard = + tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default()); + let _subscriber_guard = tracing::subscriber::set_default(subscriber); + let (session, turn, events) = make_session_and_context_with_rx().await; + session + .spawn_task( + Arc::clone(&turn), + Vec::new(), + HeldStepTask { + kind: TaskKind::Regular, + finish: Arc::new(Notify::new()), + }, + ) + .await; + turn.turn_timing_state + .mark_turn_started(std::time::Instant::now()) + .await; + drop(turn.turn_timing_state.begin_sampling()); + turn.turn_timing_state.record_sampling_retry(); + // Guardian uses this detached abort path rather than submitting Op::Interrupt. + session + .interrupt_turn_with_warning( + &turn.sub_id, + EventMsg::Warning(codex_protocol::protocol::WarningEvent { + message: "Guardian denial limit reached".into(), + }), + /*error*/ None, + ) + .instrument(tracing::info_span!("guardian_review")) + .await; + tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + while !matches!(events.recv().await?.msg, EventMsg::TurnAborted(_)) {} + anyhow::Ok(()) + }) + .await??; + + let profile = tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + loop { + let spans = exported.get_finished_spans()?; + for span in spans { + if span.name == "guardian_review" + && let Some(event) = span + .events + .iter() + .find(|event| event.name == "codex.turn_profile") + { + return anyhow::Ok(event.clone()); + } + } + tokio::task::yield_now().await; + } + }) + .await??; + let attributes: std::collections::BTreeMap<_, _> = profile + .attributes + .iter() + .filter(|attribute| { + matches!( + attribute.key.as_str(), + "conversation.id" | "turn_id" | "sampling_request_count" | "sampling_retry_count" + ) + }) + .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) + .collect(); + assert_eq!( + attributes, + std::collections::BTreeMap::from([ + ("conversation.id", session.thread_id.to_string()), + ("turn_id", turn.sub_id.clone()), + ("sampling_request_count", "1".into()), + ("sampling_retry_count", "1".into()), + ]) + ); + Ok(()) +} #[tokio::test] async fn interrupt_if_no_pending_input_handles_cancelled_submission() { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 639a09ddd0..2ebfcc173d 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -833,6 +833,11 @@ impl Session { .turn_timing_state .complete_profile_and_duration_ms() .await; + crate::turn_timing::record_turn_profile_trace( + self.thread_id, + &turn_context.sub_id, + &profile, + ); self.services .analytics_events_client .track_turn_profile(TurnProfileFact { @@ -1036,6 +1041,11 @@ impl Session { .turn_timing_state .complete_profile_and_duration_ms() .await; + crate::turn_timing::record_turn_profile_trace( + self.thread_id, + &task.turn_context.sub_id, + &profile, + ); self.services .analytics_events_client .track_turn_profile(TurnProfileFact { diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index e33999d697..3472d95953 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -44,7 +44,11 @@ impl SessionTask for RegularTask { input: Vec, cancellation_token: CancellationToken, ) -> SessionTaskResult { - let run_turn_span = trace_span!("run_turn"); + let run_turn_span = trace_span!( + "run_turn", + conversation.id = %sess.thread_id, + turn.id = %ctx.sub_id, + ); // Regular turns emit `TurnStarted` inline so first-turn lifecycle does // not wait on startup prewarm resolution. let prewarmed_client_session = async { diff --git a/codex-rs/core/src/tools/call_trace.rs b/codex-rs/core/src/tools/call_trace.rs index b6daff424b..76a9490d5c 100644 --- a/codex-rs/core/src/tools/call_trace.rs +++ b/codex-rs/core/src/tools/call_trace.rs @@ -2,6 +2,8 @@ //! These events contain call identifiers and names, never arguments or output. //! Code-mode calls reach the broker before the dispatching turn is known. +use std::time::Duration; + use codex_protocol::DEFAULT_FUNCTION_NAMESPACE; use codex_protocol::ThreadId; use codex_tools::ToolName; @@ -13,7 +15,7 @@ pub(crate) enum Source { } impl Source { - fn as_str(self) -> &'static str { + pub(crate) fn as_str(self) -> &'static str { match self { Self::Direct => "direct", Self::CodeMode => "code_mode", @@ -58,18 +60,24 @@ pub(crate) fn received( ); } +// Record the measured duration on the terminal event so a retained completion +// remains useful even when its receipt event or dispatch span was not retained. +// This includes runtime admission and handler work, but excludes broker queuing +// before dispatch and the sampling loop's later result collection. pub(crate) fn result_ready( thread_id: ThreadId, turn_id: &str, tool_name: &ToolName, call_id: &str, source: Source, + dispatch_to_result_ready: Duration, ) { tracing::event!( name: "codex.tool_result_ready", target: "codex_otel.trace_safe", tracing::Level::INFO, event.name = "codex.tool_result_ready", + dispatch_to_result_ready_ms = dispatch_to_result_ready.as_secs_f64() * 1000.0, conversation.id = %thread_id, turn_id, call_id, @@ -79,7 +87,7 @@ pub(crate) fn result_ready( ); } -fn namespace(tool_name: &ToolName) -> &str { +pub(crate) fn namespace(tool_name: &ToolName) -> &str { tool_name .namespace .as_deref() diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 2b241ffc7d..7c4172d9e5 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -345,6 +345,7 @@ fn submit_nested_tool( impl std::future::Future> + Send + 'static, FunctionCallError, > { + let dispatch_started = std::time::Instant::now(); let CodeModeNestedToolCall { cell_id, runtime_tool_call_id, @@ -361,6 +362,9 @@ fn submit_nested_tool( target: "codex_otel.trace_safe", tracing::Level::INFO, event.name = "codex.code_mode.nested_tool_dispatched", + tool_name = tool_name.name.as_str(), + tool_namespace = call_trace::namespace(&tool_name), + tool_source = "code_mode", conversation.id = %thread_id, turn_id = turn_id.as_str(), cell.id = telemetry::trace_id(cell_id.as_str()), @@ -381,6 +385,7 @@ fn submit_nested_tool( &tool_name, &call_id, call_trace::Source::CodeMode, + dispatch_started.elapsed(), ); return Err(FunctionCallError::RespondToModel(error)); } diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 1ab879940b..4dea6384f3 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -194,7 +194,10 @@ impl ToolCallRuntime { otel.name = %call.tool_name, tool_name = %call.tool_name, thread.id = %session.thread_id, + conversation.id = %session.thread_id, + turn.id = %turn.sub_id, turn_id = %turn.sub_id, + tool_source = trace_source.as_str(), call_id = call.call_id.as_str(), codex.tool.phase = "dispatch", aborted = false, @@ -259,6 +262,7 @@ impl ToolCallRuntime { &dispatch_tool_name, &dispatch_call_id, trace_source, + started.elapsed(), ); } result @@ -291,6 +295,7 @@ impl ToolCallRuntime { &call.tool_name, &call.call_id, trace_source, + started.elapsed(), ); call_state.terminal_outcome_reached.store(true, Ordering::Release); notify_tool_aborted( diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index 75a2bf9bf0..37d238b7ec 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -8,6 +8,7 @@ use std::time::UNIX_EPOCH; use codex_analytics::TurnProfile; use codex_otel::TURN_TTFM_DURATION_METRIC; +use codex_protocol::ThreadId; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseItem; use tokio::sync::Mutex; @@ -16,6 +17,22 @@ use crate::ResponseEvent; use crate::session::turn_context::TurnContext; use crate::stream_events_utils::raw_assistant_output_text_from_item; +// Reuse the finalized analytics counters instead of counting sampled child spans. +// Recovery can complete the same turn ID again, so consumers must retain the +// completion event's trace and timestamp rather than deduplicating by turn alone. +pub(crate) fn record_turn_profile_trace(thread_id: ThreadId, turn_id: &str, profile: &TurnProfile) { + tracing::event!( + name: "codex.turn_profile", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.turn_profile", + conversation.id = %thread_id, + turn_id, + sampling_request_count = profile.sampling_request_count, + sampling_retry_count = profile.sampling_retry_count, + ); +} + pub(crate) async fn record_turn_ttft_metric(turn_context: &TurnContext, event: &ResponseEvent) { let Some(duration) = turn_context .turn_timing_state diff --git a/codex-rs/core/tests/suite/turn_phase_trace.rs b/codex-rs/core/tests/suite/turn_phase_trace.rs index c242df1490..0d1d2f18f7 100644 --- a/codex-rs/core/tests/suite/turn_phase_trace.rs +++ b/codex-rs/core/tests/suite/turn_phase_trace.rs @@ -269,7 +269,22 @@ async fn steer_joins_existing_turn_without_extending_sampling( }) }) .await?; + let run_turn = wait_for_span(&exported, |span| span.name == "run_turn").await?; let spans = exported.get_finished_spans()?; + let dispatch = spans + .iter() + .find(|span| { + attribute(&span.attributes, "call_id").as_deref() == Some(call.call_id.as_str()) + && attribute(&span.attributes, "tool_source").as_deref() == Some("direct") + }) + .expect("tool dispatch can be joined without following parent spans"); + assert_eq!( + [&run_turn.attributes, &dispatch.attributes].map(|attributes| ( + attribute(attributes, "conversation.id"), + attribute(attributes, "turn.id"), + )), + std::array::from_fn::<_, 2, _>(|_| (Some(thread_id.clone()), Some(call.turn_id.clone()))) + ); let steer = spans .iter() .find(|span| span.name == "steer_request") @@ -315,7 +330,42 @@ async fn steer_joins_existing_turn_without_extending_sampling( && attribute(&event.attributes, "call_id").as_deref() == Some(call.call_id.as_str()) }) .expect("gated tool result-ready event exported"); + let dispatch_to_ready_ms = attribute(&ready.attributes, "dispatch_to_result_ready_ms") + .expect("completion carries its own timing") + .parse::()?; + assert!(dispatch_to_ready_ms.is_finite() && dispatch_to_ready_ms >= 0.0); assert!(accepted.end_time <= ready.timestamp); assert!(ready.timestamp <= blocking.end_time); + let mut profiles: Vec<_> = spans + .iter() + .flat_map(|span| span.events.events.iter()) + .filter(|event| { + event.name == "codex.turn_profile" + && attribute(&event.attributes, "turn_id").as_deref() == Some(call.turn_id.as_str()) + }) + .map(|event| { + ( + attribute(&event.attributes, "conversation.id"), + attribute(&event.attributes, "sampling_request_count"), + attribute(&event.attributes, "sampling_retry_count"), + ) + }) + .collect(); + profiles.sort(); + let expected_counts = match finish { + Finish::Reply => vec!["2"], + Finish::Interrupt => vec!["1", "1"], + }; + assert_eq!( + profiles, + expected_counts + .into_iter() + .map(|count| ( + Some(thread_id.clone()), + Some(count.to_string()), + Some("0".to_string()) + )) + .collect::>() + ); Ok(()) } diff --git a/codex-rs/ext/skills/src/tools/mod.rs b/codex-rs/ext/skills/src/tools/mod.rs index 3faff78002..ea89558116 100644 --- a/codex-rs/ext/skills/src/tools/mod.rs +++ b/codex-rs/ext/skills/src/tools/mod.rs @@ -74,6 +74,7 @@ pub(crate) fn skill_tools( .and_then(|state| state.mcp_resources.clone()); let analytics = SkillAnalytics::from_stores(session_store, thread_store); let context = SkillToolContext { + thread_id: thread_store.level_id().to_string(), providers, mcp_resources, thread_state, @@ -204,6 +205,7 @@ impl SkillAnalytics { #[derive(Clone)] struct SkillToolContext { + thread_id: String, providers: SkillProviders, mcp_resources: Option>, thread_state: Arc, diff --git a/codex-rs/ext/skills/src/tools/read.rs b/codex-rs/ext/skills/src/tools/read.rs index 37d4251ee1..a8dcaa19e6 100644 --- a/codex-rs/ext/skills/src/tools/read.rs +++ b/codex-rs/ext/skills/src/tools/read.rs @@ -23,6 +23,7 @@ use super::InstallationLocation; use super::InstallationSource; use super::MAX_HANDLE_BYTES; use super::MAX_SKILL_RESPONSE_BYTES; +use super::SKILLS_NAMESPACE; use super::SkillToolAuthority; use super::SkillToolAuthoritySelector; use super::SkillToolContext; @@ -161,6 +162,32 @@ impl<'call> ToolExecutor> for ReadTool { .bind_environment_package_resource(&package, resource.clone()) .unwrap_or_else(|| SkillResourceId::new(resource)), }; + // Record the selected route before either cache or provider can serve the read. + // This is an attempted read, not a successful invocation or a backend RPC. + tracing::event!( + name: "codex.skill_read.source_selected", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.skill_read.source_selected", + conversation.id = %self.context.thread_id, + turn_id = %call.turn_id, + call_id = %call.call_id, + tool_namespace = SKILLS_NAMESPACE, + tool_name = TOOL_NAME, + skill_source = match output_authority { + SkillToolAuthoritySelector::Cloud => "cloud", + SkillToolAuthoritySelector::Executor => "executor", + }, + skill_name = %skill_entry.name, + skill_plugin_id = selected_source.plugin_id.as_deref(), + skill_resource_kind = if requested_resource == main_prompt { + "main_prompt" + } else { + "supporting_resource" + }, + is_continuation = args.cursor.is_some(), + "Selected skill read source" + ); let (context, access_key) = match output_authority { SkillToolAuthoritySelector::Cloud => ( SkillReadContext::Cloud { diff --git a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs index 24a3ab930b..83515a44ca 100644 --- a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs +++ b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs @@ -13,6 +13,9 @@ use pretty_assertions::assert_eq; use serde_json::Value; use serde_json::json; +#[path = "source_telemetry_tests.rs"] +mod source_telemetry_tests; + fn cloud_skill(name: &str) -> SkillCatalogEntry { let package = format!("skill://Plugin_123/{name}"); SkillCatalogEntry::new( diff --git a/codex-rs/ext/skills/tests/skills_extension/source_telemetry_tests.rs b/codex-rs/ext/skills/tests/skills_extension/source_telemetry_tests.rs new file mode 100644 index 0000000000..bed10e5c45 --- /dev/null +++ b/codex-rs/ext/skills/tests/skills_extension/source_telemetry_tests.rs @@ -0,0 +1,116 @@ +//! Verifies provider attribution follows the selected route, including cached reads. + +use super::*; +use pretty_assertions::assert_eq; +use tracing::field::Field; +use tracing::field::Visit; +use tracing::instrument::WithSubscriber; +use tracing_subscriber::Layer; +use tracing_subscriber::layer::Context; +use tracing_subscriber::prelude::*; + +#[derive(Clone, Default)] +struct ReadEvents(Arc>>); + +#[derive(Default)] +struct Fields(serde_json::Map); + +impl Visit for Fields { + fn record_str(&mut self, field: &Field, value: &str) { + self.0.insert(field.name().into(), json!(value)); + } + + fn record_bool(&mut self, field: &Field, value: bool) { + self.0.insert(field.name().into(), json!(value)); + } + + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + self.record_str(field, &format!("{value:?}")); + } +} + +impl Layer for ReadEvents { + fn on_event(&self, event: &tracing::Event<'_>, _context: Context<'_, S>) { + let mut fields = Fields::default(); + event.record(&mut fields); + if fields.0.get("event.name") == Some(&json!("codex.skill_read.source_selected")) { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .push(Value::Object(fields.0)); + } + } +} + +#[tokio::test] +async fn source_events_cover_cached_cloud_and_explicit_executor_reads() -> TestResult { + let cloud = cloud_skill("work").with_plugin_id(Some("cloud-plugin".into())); + let installed = installed_skill("matching", "work")?; + let cloud_provider = provider(vec![cloud.clone()]); + let cloud_reads = cloud_provider.read_requests.clone(); + let executor = provider(vec![installed]); + let executor_reads = executor.read_requests.clone(); + let harness = MergeHarness::new(executor, cloud_provider, default_config()).await?; + harness + .turn_store + .insert(selected_plugins(&[None, Some("env-1:matching")], &[])); + let tools = harness.tools(&harness.step_store).await; + harness.render().await?; + let events = ReadEvents::default(); + let subscriber = tracing_subscriber::registry().with(events.clone().with_filter( + tracing_subscriber::filter::filter_fn(codex_otel::OtelProvider::trace_export_filter), + )); + // Prevent parallel tests with no subscriber from disabling this callsite globally. + let _interest_cache_guard = + tracing::Dispatch::new(tracing::subscriber::NoSubscriber::default()); + async { + for source in ["cloud", "cloud", "executor", "executor"] { + let selector = if source == "executor" { + json!({"kind": source, "environment_id": "env-1"}) + } else { + json!({"kind": source}) + }; + call_tool( + &tools, + "read", + json!({"package": cloud.id.0, "source": selector}), + ) + .await?; + } + assert!( + call_tool(&tools, "read", json!({"package": "missing"})) + .await + .is_err() + ); + TestResult::Ok(()) + } + .with_subscriber(subscriber) + .await?; + + let expected = ["cloud", "cloud", "executor", "executor"].map(|source| { + json!({ + "message": "Selected skill read source", + "event.name": "codex.skill_read.source_selected", + "conversation.id": "thread", + "turn_id": "turn-1", + "call_id": "skill-test", + "tool_namespace": "skills", + "tool_name": "read", + "skill_source": source, + "skill_name": "deploy:work", + "skill_plugin_id": if source == "cloud" { "cloud-plugin" } else { "matching" }, + "skill_resource_kind": "main_prompt", + "is_continuation": false, + }) + }); + assert_eq!(*events.0.lock().unwrap(), expected); + // Both cloud reads are attributable even though the second needs no provider call. + assert_eq!( + ( + read_request_keys(&cloud_reads).len(), + read_request_keys(&executor_reads).len() + ), + (1, 2), + ); + Ok(()) +}