openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-24 18:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add turn, tool, and skill attribution to traces (#56524)

Browse Source 
## What changed

- Emit `codex.turn_profile` events with finalized sampling request and retry counts when turns complete or abort, and preserve the caller's tracing span for detached extension interrupts.
- Add conversation and turn identifiers to turn and tool dispatch spans, identify tool sources, and include dispatch-to-result-ready duration on tool completion events.
- Emit `codex.skill_read.source_selected` before cache or provider access, recording the selected cloud or executor source, skill and plugin identity, resource kind, and continuation status.

## Testing

Add regression coverage for trace correlation, tool completion timing, turn profile counters on completion and interruption, detached interrupt span propagation, and skill source attribution including cached cloud reads.

GitOrigin-RevId: db78aa270bbcb8dc9e4558d202c109ad54141811
This commit is contained in:
bkotsopoulos
2026-10-23 16:55:32 +00:00
committed by copyberry
parent 3985001067
commit b82d542aef
13 changed files with 363 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/extension_interruption.rs
+17 -11
View File
@@ -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(),
),
);
}
}
codex-rs/core/src/session/extension_interruption_tests.rs
+96 -1
View File
@@ -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() {
codex-rs/core/src/tasks/mod.rs
+10
View File
@@ -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 {
codex-rs/core/src/tasks/regular.rs
+5 -1
View File
@@ -44,7 +44,11 @@ impl SessionTask for RegularTask {
input: Vec<TurnInput>,
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 {
codex-rs/core/src/tools/call_trace.rs
+10 -2
View File
@@ -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()
codex-rs/core/src/tools/code_mode/mod.rs
+5
View File
@@ -345,6 +345,7 @@ fn submit_nested_tool(
impl std::future::Future<Output = Result<JsonValue, FunctionCallError>> + 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));
}
codex-rs/core/src/tools/parallel.rs
+5
View File
@@ -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(
codex-rs/core/src/turn_timing.rs
+17
View File
@@ -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
codex-rs/core/tests/suite/turn_phase_trace.rs
+50
View File
@@ -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::<f64>()?;
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::<Vec<_>>()
);
Ok(())
}
codex-rs/ext/skills/src/tools/mod.rs
+2
View File
@@ -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<Arc<McpResourceClient>>,
thread_state: Arc<SkillsThreadState>,
codex-rs/ext/skills/src/tools/read.rs
+27
View File
@@ -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<ToolCall<'call>> 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 {
codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs
+3
View File
@@ -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(
codex-rs/ext/skills/tests/skills_extension/source_telemetry_tests.rs
+116
View File
@@ -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<Mutex<Vec<Value>>>);
#[derive(Default)]
struct Fields(serde_json::Map<String, Value>);
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<S: tracing::Subscriber> Layer<S> 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(())
}