diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index 8dc8f359d4..8f6c61fdee 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -81,6 +81,12 @@ pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_BYTES: usize = 1024 * 1024; // 1 MiB pub(crate) const UNIFIED_EXEC_OUTPUT_MAX_TOKENS: usize = UNIFIED_EXEC_OUTPUT_MAX_BYTES / 4; pub(crate) const MAX_UNIFIED_EXEC_PROCESSES: usize = 64; +const MAX_TRACE_ID_BYTES: usize = 256; + +fn trace_id(id: &str) -> Option<&str> { + (!id.is_empty() && id.len() <= MAX_TRACE_ID_BYTES).then_some(id) +} + pub(crate) struct UnifiedExecContext { pub session: Arc, pub step_context: Arc, diff --git a/codex-rs/core/src/unified_exec/oneshot.rs b/codex-rs/core/src/unified_exec/oneshot.rs index ec7cffdb16..460aaeb462 100644 --- a/codex-rs/core/src/unified_exec/oneshot.rs +++ b/codex-rs/core/src/unified_exec/oneshot.rs @@ -8,12 +8,14 @@ use std::sync::OnceLock; use tokio::time::Duration; use tokio::time::Instant; +use tracing::Instrument; use super::ExecCommandRequest; use super::UnifiedExecContext; use super::UnifiedExecError; use super::UnifiedExecProcess; use super::UnifiedExecProcessManager; +use super::trace_id; use crate::tools::context::ExecCommandToolOutput; pub(super) struct Completion<'a> { @@ -23,6 +25,19 @@ pub(super) struct Completion<'a> { } impl UnifiedExecProcessManager { + #[tracing::instrument( + name = "unified_exec.exec_command", + level = "info", + skip_all, + fields( + conversation.id = %context.session.thread_id, + turn_id = trace_id(&context.step_context.turn.sub_id), + call_id = trace_id(&context.call_id), + unified_exec_process_id = request.process_id, + mode = "oneshot", + outcome = tracing::field::Empty, + ) + )] pub(crate) async fn exec_command_to_completion( request: ExecCommandRequest, context: &UnifiedExecContext, @@ -35,14 +50,17 @@ impl UnifiedExecProcessManager { context.call_id.clone(), ); let _cancel_on_drop = context.cancellation_token.clone().drop_guard(); - tokio::spawn(async move { + let task = async move { let manager = &context.session.services.unified_exec_manager; let process_id = request.process_id; if Instant::now().checked_add(timeout).is_none() { manager.release_process_id(process_id).await; - return Err(UnifiedExecError::process_failed( - "timeout_ms is too large".into(), - )); + return ( + Err(UnifiedExecError::process_failed( + "timeout_ms is too large".into(), + )), + "failed", + ); } let process = OnceLock::new(); @@ -68,20 +86,38 @@ impl UnifiedExecProcessManager { drop(execution); manager.release_process_id(process_id).await; } - Err(UnifiedExecError::process_failed("command cancelled".into())) + return ( + Err(UnifiedExecError::process_failed("command cancelled".into())), + "cancelled", + ); } result = &mut execution => result, } }; - result.map(|mut output| { + let outcome = if completion.timed_out { + "timed_out" + } else if result.is_ok() { + "exited" + } else { + "failed" + }; + let result = result.map(|mut output| { if completion.timed_out { output.process_id = None; } output - }) - }) - .await - .map_err(|err| UnifiedExecError::process_failed(err.to_string()))? + }); + (result, outcome) + }; + let (result, outcome) = match tokio::spawn(task.in_current_span()).await { + Ok(result) => result, + Err(err) => ( + Err(UnifiedExecError::process_failed(err.to_string())), + "failed", + ), + }; + tracing::Span::current().record("outcome", outcome); + result } } diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index 676acd9f96..e3be5f38e0 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -67,6 +67,7 @@ use crate::unified_exec::process::SpawnLifecycleHandle; use crate::unified_exec::process::UnifiedExecProcess; use crate::unified_exec::shell_snapshot::shell_snapshot_request; use crate::unified_exec::take_plugin_metrics_sidecar; +use crate::unified_exec::trace_id; use codex_core_plugins::PLUGIN_METRICS_OUTPUT_ENV_VAR; use codex_core_plugins::PluginCommandAttribution; use codex_core_plugins::PluginMetricsSidecar; @@ -487,13 +488,35 @@ impl UnifiedExecProcessManager { } } + #[tracing::instrument( + name = "unified_exec.exec_command", + level = "info", + skip_all, + fields( + conversation.id = %context.session.thread_id, + turn_id = trace_id(&context.step_context.turn.sub_id), + call_id = trace_id(&context.call_id), + unified_exec_process_id = request.process_id, + mode = "resumable", + outcome = tracing::field::Empty, + ) + )] pub(crate) async fn exec_command( &self, request: ExecCommandRequest, context: &UnifiedExecContext, ) -> Result { - self.exec_command_inner(request, context, /*completion*/ None) - .await + let result = self + .exec_command_inner(request, context, /*completion*/ None) + .await; + let outcome = match &result { + Ok(output) if output.process_id.is_some() => "yielded", + Ok(_) => "exited", + Err(_) if context.cancellation_token.is_cancelled() => "cancelled", + Err(_) => "failed", + }; + tracing::Span::current().record("outcome", outcome); + result } pub(super) async fn exec_command_inner( @@ -819,10 +842,40 @@ impl UnifiedExecProcessManager { Ok(response) } + #[tracing::instrument( + name = "unified_exec.write_stdin", + level = "info", + skip_all, + fields( + conversation.id = %context.session.thread_id, + turn_id = trace_id(&context.step_context.turn.sub_id), + call_id = trace_id(&context.call_id), + original_exec_call_id = tracing::field::Empty, + unified_exec_process_id = request.process_id, + interaction = if request.input.is_empty() { "poll" } else { "write" }, + outcome = tracing::field::Empty, + ) + )] pub(crate) async fn write_stdin( &self, context: &UnifiedExecContext, request: WriteStdinRequest<'_>, + ) -> Result { + let result = self.write_stdin_inner(context, request).await; + let outcome = match &result { + Ok(output) if output.process_id.is_some() => "yielded", + Ok(_) => "exited", + Err(_) if context.cancellation_token.is_cancelled() => "cancelled", + Err(_) => "failed", + }; + tracing::Span::current().record("outcome", outcome); + result + } + + async fn write_stdin_inner( + &self, + context: &UnifiedExecContext, + request: WriteStdinRequest<'_>, ) -> Result { let process_id = request.process_id; @@ -835,6 +888,10 @@ impl UnifiedExecProcessManager { .processes .get(&process_id) .ok_or(UnifiedExecError::UnknownProcessId { process_id })?; + // Capture the original call even if this interaction is cancelled while queued. + if let Some(call_id) = trace_id(&entry.call_id) { + tracing::Span::current().record("original_exec_call_id", call_id); + } Arc::clone(&entry.process) }; let _interaction_guard = locked_process.interaction_lock().lock_owned().await; @@ -1278,6 +1335,15 @@ impl UnifiedExecProcessManager { windows_sandbox_proxy_settings_mode, tty, ); + // Sandbox retries can reuse the public ID for a new executor process. + tracing::event!( + name: "codex.unified_exec.process_start_requested", + target: "codex_otel.trace_safe", + tracing::Level::INFO, + event.name = "codex.unified_exec.process_start_requested", + unified_exec_process_id = process_id, + process.id = params.process_id.as_str(), + ); let started = match network_policy_decider { Some(decider) => { backend @@ -1364,6 +1430,12 @@ impl UnifiedExecProcessManager { UnifiedExecProcess::from_spawned(spawned, request.sandbox, spawn_lifecycle).await } + #[tracing::instrument( + name = "unified_exec.open_session", + level = "info", + skip_all, + fields(outcome = tracing::field::Empty) + )] pub(super) async fn open_session_with_sandbox( &self, request: &ExecCommandRequest, @@ -1458,7 +1530,7 @@ impl UnifiedExecProcessManager { call_id: context.call_id.clone(), tool_name: ToolName::plain("exec_command"), }; - orchestrator + let result = orchestrator .run(&mut runtime, &req, &tool_ctx) .await .map(|result| (result.output, result.deferred_network_approval)) @@ -1477,9 +1549,27 @@ impl UnifiedExecProcessManager { _ => UnifiedExecError::create_process(format!("{err:?}")), }, other => UnifiedExecError::create_process(format!("{other:?}")), - }) + }); + let outcome = match &result { + Ok(_) => "completed", + Err(_) if context.cancellation_token.is_cancelled() => "cancelled", + Err(_) => "failed", + }; + tracing::Span::current().record("outcome", outcome); + result } + #[tracing::instrument( + name = "unified_exec.collect_output", + level = "info", + skip_all, + fields( + outcome = tracing::field::Empty, + stop_reason = tracing::field::Empty, + exit_signaled = tracing::field::Empty, + output_closed = tracing::field::Empty, + ) + )] pub(super) async fn collect_output_until_deadline( output: &OutputHandles, mut pause_state: Option>, @@ -1487,6 +1577,12 @@ impl UnifiedExecProcessManager { ) -> HeadTailBuffer { const POST_EXIT_CLOSE_WAIT_CAP: Duration = Duration::from_millis(50); + enum StopReason { + OutputClosed, + Deadline, + PostExitDeadline, + } + let OutputHandles { output_buffer, output_notify, @@ -1497,7 +1593,7 @@ impl UnifiedExecProcessManager { let mut collected = HeadTailBuffer::default(); let mut exit_signal_received = cancellation_token.is_cancelled(); let mut post_exit_deadline: Option = None; - loop { + let stop_reason = loop { Self::extend_deadlines_while_paused( &mut pause_state, &mut deadline, @@ -1519,13 +1615,12 @@ impl UnifiedExecProcessManager { if !has_drained_output { exit_signal_received |= cancellation_token.is_cancelled(); - if exit_signal_received && output_closed.load(std::sync::atomic::Ordering::Acquire) - { - break; + if exit_signal_received && output_closed.load(Ordering::Acquire) { + break StopReason::OutputClosed; } let remaining = deadline.saturating_duration_since(Instant::now()); if remaining == Duration::ZERO { - break; + break StopReason::Deadline; } if exit_signal_received { @@ -1534,7 +1629,7 @@ impl UnifiedExecProcessManager { .get_or_insert_with(|| now + remaining.min(POST_EXIT_CLOSE_WAIT_CAP)); let close_wait_remaining = close_wait_deadline.saturating_duration_since(now); if close_wait_remaining == Duration::ZERO { - break; + break StopReason::PostExitDeadline; } let notified = wait_for_output.unwrap_or_else(|| output_notify.notified()); let closed = output_closed_notify.notified(); @@ -1543,7 +1638,7 @@ impl UnifiedExecProcessManager { tokio::select! { _ = &mut notified => {} _ = &mut closed => {} - _ = tokio::time::sleep(close_wait_remaining) => break, + _ = tokio::time::sleep(close_wait_remaining) => break StopReason::PostExitDeadline, _ = Self::wait_for_pause_change(pause_state.as_ref()) => {} } continue; @@ -1556,7 +1651,7 @@ impl UnifiedExecProcessManager { tokio::select! { _ = &mut notified => {} _ = &mut exit_notified => exit_signal_received = true, - _ = tokio::time::sleep(remaining) => break, + _ = tokio::time::sleep(remaining) => break StopReason::Deadline, _ = Self::wait_for_pause_change(pause_state.as_ref()) => {} } continue; @@ -1566,10 +1661,25 @@ impl UnifiedExecProcessManager { exit_signal_received |= cancellation_token.is_cancelled(); if Instant::now() >= deadline { - break; + break StopReason::Deadline; } - } + }; + let span = tracing::Span::current(); + span.record( + "stop_reason", + match stop_reason { + StopReason::OutputClosed => "output_closed", + StopReason::Deadline => "deadline", + StopReason::PostExitDeadline => "post_exit_deadline", + }, + ); + span.record( + "exit_signaled", + exit_signal_received || cancellation_token.is_cancelled(), + ); + span.record("output_closed", output_closed.load(Ordering::Acquire)); + span.record("outcome", "completed"); collected }