openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add lifecycle tracing for unified exec (#45505)

Browse Source 
## What changed

- Add spans for one-shot and resumable `exec_command`, `write_stdin`, session creation, and output collection, recording outcomes and output collection stop reasons.
- Correlate calls with conversations, turns, and processes; link stdin interactions to the original exec call and process start requests to executor process IDs. Omit empty turn and call IDs and those longer than 256 bytes.
- Propagate the current tracing span into the spawned one-shot execution task and distinguish timeouts, cancellations, and failures.

GitOrigin-RevId: 722728dc3f5b624e7a4da69fc867c3c672465af0
This commit is contained in:
Adam Perry @ OpenAI
2026-09-14 19:23:36 +00:00
committed by copyberry
parent 6ce16aadce
commit 4d5d37c5f8
3 changed files with 176 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
codex-rs/core/src/unified_exec/mod.rs
View File

@@ -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<Session>,
pub step_context: Arc<StepContext>,

56
codex-rs/core/src/unified_exec/oneshot.rs
View File

@@ -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
}
}

138
codex-rs/core/src/unified_exec/process_manager.rs
View File

@@ -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<ExecCommandToolOutput, UnifiedExecError> {
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<ExecCommandToolOutput, UnifiedExecError> {
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<ExecCommandToolOutput, UnifiedExecError> {
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<const MAX_BYTES: usize>(
output: &OutputHandles<MAX_BYTES>,
mut pause_state: Option<watch::Receiver<bool>>,
@@ -1487,6 +1577,12 @@ impl UnifiedExecProcessManager {
) -> HeadTailBuffer<MAX_BYTES> {
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<Instant> = 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
}