From eb06bdcc87af37f1fada6ea8fbc109453c6468e5 Mon Sep 17 00:00:00 2001 From: Anton Panasenko Date: Mon, 29 Jun 2026 22:09:35 -0700 Subject: [PATCH] telemetry(core): trace unified exec process lifecycle --- .../core/src/unified_exec/async_watcher.rs | 7 + codex-rs/core/src/unified_exec/process.rs | 432 ++++++++++++------ .../core/src/unified_exec/process_manager.rs | 185 ++++++-- .../core/src/unified_exec/process_tests.rs | 31 ++ 4 files changed, 486 insertions(+), 169 deletions(-) diff --git a/codex-rs/core/src/unified_exec/async_watcher.rs b/codex-rs/core/src/unified_exec/async_watcher.rs index 38010754e9..81d00faf43 100644 --- a/codex-rs/core/src/unified_exec/async_watcher.rs +++ b/codex-rs/core/src/unified_exec/async_watcher.rs @@ -5,6 +5,7 @@ use tokio::sync::Mutex; use tokio::time::Duration; use tokio::time::Instant; use tokio::time::Sleep; +use tracing::instrument; use super::UnifiedExecContext; use super::process::UnifiedExecProcess; @@ -192,6 +193,7 @@ async fn process_chunk( /// as the primary source of aggregated_output and falling back to the provided /// text when the transcript is empty. #[allow(clippy::too_many_arguments)] +#[instrument(name = "unified_exec.emit_end", level = "info", skip_all)] pub(crate) async fn emit_exec_end_for_unified_exec( session_ref: Arc, turn_ref: Arc, @@ -317,6 +319,11 @@ fn split_valid_utf8_prefix_with_max(buffer: &mut Vec, max_bytes: usize) -> O Some(byte) } +#[instrument( + name = "unified_exec.resolve_aggregated_output", + level = "info", + skip_all +)] async fn resolve_aggregated_output( transcript: &Arc>, fallback: String, diff --git a/codex-rs/core/src/unified_exec/process.rs b/codex-rs/core/src/unified_exec/process.rs index 7bedfd43e3..458da15972 100644 --- a/codex-rs/core/src/unified_exec/process.rs +++ b/codex-rs/core/src/unified_exec/process.rs @@ -10,7 +10,11 @@ use tokio::sync::oneshot::error::TryRecvError; use tokio::sync::watch; use tokio::task::JoinHandle; use tokio::time::Duration; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use tracing::Instrument; +use tracing::field; +use tracing::instrument; use crate::exec::is_likely_sandbox_denied; use codex_exec_server::ExecProcess; @@ -266,17 +270,33 @@ impl UnifiedExecProcess { self.state_rx.borrow().failure_message.clone() } + #[instrument( + name = "unified_exec.check_for_sandbox_denial", + level = "info", + skip_all + )] pub(super) async fn check_for_sandbox_denial(&self) -> Result<(), UnifiedExecError> { - let _ = - tokio::time::timeout(Duration::from_millis(20), self.output_notify.notified()).await; + let _ = tokio::time::timeout(Duration::from_millis(20), self.output_notify.notified()) + .instrument(tracing::info_span!( + "unified_exec.check_for_sandbox_denial.wait_for_output" + )) + .await; - let collected_chunks = self.snapshot_output().await; + let collected_chunks = self + .snapshot_output() + .instrument(tracing::info_span!( + "unified_exec.check_for_sandbox_denial.snapshot_output" + )) + .await; let mut aggregated: Vec = Vec::new(); for chunk in collected_chunks { aggregated.extend_from_slice(&chunk); } let aggregated_text = String::from_utf8_lossy(&aggregated).to_string(); self.check_for_sandbox_denial_with_text(&aggregated_text) + .instrument(tracing::info_span!( + "unified_exec.check_for_sandbox_denial.classify" + )) .await?; Ok(()) @@ -374,8 +394,29 @@ impl UnifiedExecProcess { Ok(managed) } + #[instrument( + name = "unified_exec.attach_exec_server_process", + level = "info", + skip_all + )] pub(super) async fn from_exec_server_started( started: StartedExecProcess, + ) -> Result { + Self::from_exec_server_started_with_output_task( + started, + Self::spawn_exec_server_output_task, + ) + .await + } + + pub(super) async fn from_exec_server_started_with_output_task( + started: StartedExecProcess, + spawn_output_task: impl FnOnce( + StartedExecProcess, + OutputHandles, + broadcast::Sender>, + watch::Sender, + ) -> JoinHandle<()>, ) -> Result { let process_handle = ProcessHandle::ExecServer(Arc::clone(&started.process)); let mut managed = Self::new( @@ -384,15 +425,24 @@ impl UnifiedExecProcess { /*spawn_lifecycle*/ None, ); let output_handles = managed.output_handles(); - managed.output_task = Some(Self::spawn_exec_server_output_task( + managed.output_task = Some(spawn_output_task( started, output_handles, managed.output_tx.clone(), managed.state_tx.clone(), )); + // Keep the initial sandbox-denial handshake inside ToolOrchestrator's attempt. A denial + // discovered only by exec_command's later output collection is too late for the + // orchestrator to perform its approval/escalation retry. This remains event-driven and + // returns as soon as the exec-server reports a terminal launch state. let mut state_rx = managed.state_rx.clone(); - if tokio::time::timeout(EARLY_EXIT_GRACE_PERIOD, async { + let terminal_handshake_span = tracing::info_span!( + "unified_exec.initial_terminal_handshake", + timeout_ms = EARLY_EXIT_GRACE_PERIOD.as_millis(), + completion_reason = field::Empty, + ); + let terminal_handshake = tokio::time::timeout(EARLY_EXIT_GRACE_PERIOD, async { loop { let state = state_rx.borrow().clone(); if state.has_exited || state.failure_message.is_some() { @@ -403,9 +453,17 @@ impl UnifiedExecProcess { } } }) - .await - .is_ok() - { + .instrument(terminal_handshake_span.clone()) + .await; + terminal_handshake_span.record( + "completion_reason", + if terminal_handshake.is_ok() { + "terminal_state" + } else { + "timeout" + }, + ); + if terminal_handshake.is_ok() { managed.check_for_sandbox_denial().await?; } @@ -426,153 +484,249 @@ impl UnifiedExecProcess { cancellation_token, } = output_handles; let process = started.process; + let process_id = process.process_id().to_string(); let mut events = process.subscribe_events(); - tokio::spawn(async move { - let mut last_seq: u64 = 0; - loop { - let event = match events.recv().await { - Ok(event) => Some(event), - Err(broadcast::error::RecvError::Lagged(_)) => None, - Err(broadcast::error::RecvError::Closed) => { - let state = state_tx.borrow().clone(); - let _ = state_tx.send_replace( - state.failed("exec-server process event stream closed".to_string()), - ); - output_closed.store(true, Ordering::Release); - output_closed_notify.notify_waiters(); - cancellation_token.cancel(); - break; - } - }; - let event_seq = event.as_ref().and_then(|event| match event { - ExecProcessEvent::Output(chunk) => Some(chunk.seq), - ExecProcessEvent::Exited { seq, .. } | ExecProcessEvent::Closed { seq } => { - Some(*seq) - } - ExecProcessEvent::Failed(_) => None, - }); - let missing_sandbox_denial = matches!( - event.as_ref(), - Some(ExecProcessEvent::Exited { - sandbox_denied: None, - .. - }) - ); - if event.is_none() - || event_seq.is_some_and(|seq| seq > last_seq.saturating_add(1)) - || missing_sandbox_denial - { - let response = match process - .read( - Some(last_seq), - /*max_bytes*/ None, - /*wait_ms*/ Some(0), - ) - .await - { - Ok(response) => response, - Err(err) => { + let lifecycle_started_at = Instant::now(); + let lifecycle_span = tracing::info_span!( + "exec_server.remote.process_lifecycle", + process_id = %process_id, + attach_to_first_output_ms = field::Empty, + first_output_seq = field::Empty, + first_output_bytes = field::Empty, + attach_to_exit_ms = field::Empty, + exit_code = field::Empty, + attach_to_closed_ms = field::Empty, + recovery_reads = 0_u64, + completion_reason = field::Empty, + ); + tokio::spawn( + async move { + let mut last_seq: u64 = 0; + let mut first_output_observed = false; + let mut exit_observed = false; + let mut recovery_reads: u64 = 0; + loop { + let event = match events.recv().await { + Ok(event) => Some(event), + Err(broadcast::error::RecvError::Lagged(_)) => None, + Err(broadcast::error::RecvError::Closed) => { + tracing::Span::current() + .record("completion_reason", "event_stream_closed"); let state = state_tx.borrow().clone(); - let _ = state_tx.send_replace(state.failed(err.to_string())); + let _ = state_tx.send_replace( + state.failed("exec-server process event stream closed".to_string()), + ); output_closed.store(true, Ordering::Release); output_closed_notify.notify_waiters(); cancellation_token.cancel(); break; } }; - let ExecReadResponse { - chunks, - next_seq, - exited, - exit_code, - closed, - failure, - sandbox_denied, - } = response; - for chunk in chunks.into_iter().filter(|chunk| chunk.seq > last_seq) { - let bytes = chunk.chunk.into_inner(); - let mut guard = output_buffer.lock().await; - guard.push_chunk(bytes.clone()); - drop(guard); - let _ = output_tx.send(bytes); - output_notify.notify_waiters(); - } - last_seq = last_seq.max(next_seq.saturating_sub(1)); - if let Some(message) = failure { - let state = state_tx.borrow().clone(); - let _ = state_tx.send_replace(state.failed(message)); - output_closed.store(true, Ordering::Release); - output_closed_notify.notify_waiters(); - cancellation_token.cancel(); - break; - } - if sandbox_denied || exited { - let mut state = state_tx.borrow().clone(); - state.sandbox_denied |= sandbox_denied; - let _ = state_tx.send_replace(if exited { - state.exited(exit_code) - } else { - state - }); - } - if closed { - output_closed.store(true, Ordering::Release); - output_closed_notify.notify_waiters(); - cancellation_token.cancel(); - break; - } - continue; - } - let Some(event) = event else { - continue; - }; - match event { - ExecProcessEvent::Output(chunk) => { - if chunk.seq <= last_seq { - continue; + let event_seq = event.as_ref().and_then(|event| match event { + ExecProcessEvent::Output(chunk) => Some(chunk.seq), + ExecProcessEvent::Exited { seq, .. } | ExecProcessEvent::Closed { seq } => { + Some(*seq) } - last_seq = chunk.seq; - let bytes = chunk.chunk.into_inner(); - let mut guard = output_buffer.lock().await; - guard.push_chunk(bytes.clone()); - drop(guard); - let _ = output_tx.send(bytes); - output_notify.notify_waiters(); - } - ExecProcessEvent::Exited { - seq, - exit_code, - sandbox_denied, - } => { - if seq <= last_seq { - continue; + ExecProcessEvent::Failed(_) => None, + }); + let missing_sandbox_denial = matches!( + event.as_ref(), + Some(ExecProcessEvent::Exited { + sandbox_denied: None, + .. + }) + ); + if event.is_none() + || event_seq.is_some_and(|seq| seq > last_seq.saturating_add(1)) + || missing_sandbox_denial + { + let recovery_reason = if event.is_none() { + "lagged" + } else if missing_sandbox_denial { + "missing_sandbox_denial" + } else { + "sequence_gap" + }; + recovery_reads += 1; + tracing::Span::current().record("recovery_reads", recovery_reads); + let response = match process + .read( + Some(last_seq), + /*max_bytes*/ None, + /*wait_ms*/ Some(0), + ) + .instrument(tracing::info_span!( + "exec_server.remote.recovery_read", + process_id = %process_id, + reason = recovery_reason, + after_seq = last_seq, + observed_event_seq = ?event_seq, + )) + .await + { + Ok(response) => response, + Err(err) => { + tracing::Span::current() + .record("completion_reason", "recovery_read_failed"); + let state = state_tx.borrow().clone(); + let _ = state_tx.send_replace(state.failed(err.to_string())); + output_closed.store(true, Ordering::Release); + output_closed_notify.notify_waiters(); + cancellation_token.cancel(); + break; + } + }; + let ExecReadResponse { + chunks, + next_seq, + exited, + exit_code, + closed, + failure, + sandbox_denied, + } = response; + for chunk in chunks.into_iter().filter(|chunk| chunk.seq > last_seq) { + let seq = chunk.seq; + let bytes = chunk.chunk.into_inner(); + if !first_output_observed { + first_output_observed = true; + let span = tracing::Span::current(); + span.record( + "attach_to_first_output_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + span.record("first_output_seq", seq); + span.record("first_output_bytes", bytes.len()); + } + let mut guard = output_buffer.lock().await; + guard.push_chunk(bytes.clone()); + drop(guard); + let _ = output_tx.send(bytes); + output_notify.notify_waiters(); } - last_seq = seq; - let mut state = state_tx.borrow().clone(); - state.sandbox_denied |= sandbox_denied.unwrap_or(false); - let _ = state_tx.send_replace(state.exited(Some(exit_code))); - } - ExecProcessEvent::Closed { seq } => { - if seq <= last_seq { - continue; + last_seq = last_seq.max(next_seq.saturating_sub(1)); + if let Some(message) = failure { + tracing::Span::current().record("completion_reason", "process_failed"); + let state = state_tx.borrow().clone(); + let _ = state_tx.send_replace(state.failed(message)); + output_closed.store(true, Ordering::Release); + output_closed_notify.notify_waiters(); + cancellation_token.cancel(); + break; } - output_closed.store(true, Ordering::Release); - output_closed_notify.notify_waiters(); - cancellation_token.cancel(); - break; + if sandbox_denied || exited { + if exited && !exit_observed { + exit_observed = true; + let span = tracing::Span::current(); + span.record( + "attach_to_exit_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + if let Some(exit_code) = exit_code { + span.record("exit_code", exit_code); + } + } + let mut state = state_tx.borrow().clone(); + state.sandbox_denied |= sandbox_denied; + let _ = state_tx.send_replace(if exited { + state.exited(exit_code) + } else { + state + }); + } + if closed { + let span = tracing::Span::current(); + span.record( + "attach_to_closed_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + span.record("completion_reason", "closed_after_recovery"); + output_closed.store(true, Ordering::Release); + output_closed_notify.notify_waiters(); + cancellation_token.cancel(); + break; + } + continue; } - ExecProcessEvent::Failed(message) => { - let state = state_tx.borrow().clone(); - let _ = state_tx.send_replace(state.failed(message)); - output_closed.store(true, Ordering::Release); - output_closed_notify.notify_waiters(); - cancellation_token.cancel(); - break; + + let Some(event) = event else { + continue; + }; + match event { + ExecProcessEvent::Output(chunk) => { + if chunk.seq <= last_seq { + continue; + } + last_seq = chunk.seq; + let bytes = chunk.chunk.into_inner(); + if !first_output_observed { + first_output_observed = true; + let span = tracing::Span::current(); + span.record( + "attach_to_first_output_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + span.record("first_output_seq", last_seq); + span.record("first_output_bytes", bytes.len()); + } + let mut guard = output_buffer.lock().await; + guard.push_chunk(bytes.clone()); + drop(guard); + let _ = output_tx.send(bytes); + output_notify.notify_waiters(); + } + ExecProcessEvent::Exited { + seq, + exit_code, + sandbox_denied, + } => { + if seq <= last_seq { + continue; + } + last_seq = seq; + if !exit_observed { + exit_observed = true; + let span = tracing::Span::current(); + span.record( + "attach_to_exit_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + span.record("exit_code", exit_code); + } + let mut state = state_tx.borrow().clone(); + state.sandbox_denied |= sandbox_denied.unwrap_or(false); + let _ = state_tx.send_replace(state.exited(Some(exit_code))); + } + ExecProcessEvent::Closed { seq } => { + if seq <= last_seq { + continue; + } + let span = tracing::Span::current(); + span.record( + "attach_to_closed_ms", + lifecycle_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + span.record("completion_reason", "closed"); + output_closed.store(true, Ordering::Release); + output_closed_notify.notify_waiters(); + cancellation_token.cancel(); + break; + } + ExecProcessEvent::Failed(message) => { + tracing::Span::current().record("completion_reason", "process_failed"); + let state = state_tx.borrow().clone(); + let _ = state_tx.send_replace(state.failed(message)); + output_closed.store(true, Ordering::Release); + output_closed_notify.notify_waiters(); + cancellation_token.cancel(); + break; + } } } } - }) + .instrument(lifecycle_span), + ) } fn spawn_local_output_task( diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index a895708f79..a7ce9d71c2 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -10,6 +10,10 @@ use tokio::sync::watch; use tokio::time::Duration; use tokio::time::Instant; use tokio_util::sync::CancellationToken; +use tracing::Instrument; +use tracing::Span; +use tracing::field; +use tracing::instrument; use uuid::Uuid; use crate::codex_thread::BackgroundTerminalInfo; @@ -253,6 +257,9 @@ async fn finish_deferred_network_approval_for_session( return Ok(()); }; finish_deferred_network_approval(session.as_ref(), deferred) + .instrument(tracing::info_span!( + "unified_exec.post_exit.finalize_network_registration" + )) .await .map_err(network_approval_error_message) } @@ -277,20 +284,36 @@ async fn network_denial_message_for_session( } } +#[instrument( + name = "unified_exec.post_exit.wait_for_late_network_denial", + level = "info", + skip_all, + fields(has_deferred = network_cancelled.is_some(), denial_observed = field::Empty) +)] async fn wait_for_late_network_denial(network_cancelled: Option) -> bool { let Some(network_cancelled) = network_cancelled else { + Span::current().record("denial_observed", false); return false; }; if network_cancelled.is_cancelled() { + Span::current().record("denial_observed", true); return true; } - tokio::select! { + let denial_observed = tokio::select! { _ = network_cancelled.cancelled() => true, _ = tokio::time::sleep(LATE_NETWORK_DENIAL_GRACE_PERIOD) => false, - } + }; + Span::current().record("denial_observed", denial_observed); + denial_observed } +#[instrument( + name = "unified_exec.post_exit.finish_network_approval", + level = "info", + skip_all, + fields(has_deferred = deferred.is_some()) +)] async fn finish_deferred_network_approval_after_process_exit_for_session( session: Option<&Arc>, deferred: Option, @@ -405,11 +428,29 @@ impl UnifiedExecProcessManager { } } + #[instrument( + name = "unified_exec.exec_command", + level = "info", + skip_all, + fields( + process_id = request.process_id, + environment_id = %request.turn_environment.environment_id, + remote = request.turn_environment.environment.is_remote(), + tty = request.tty, + output_collected_ms = field::Empty, + network_approval_complete_ms = field::Empty, + end_event_complete_ms = field::Empty, + process_released_ms = field::Empty, + sandbox_check_complete_ms = field::Empty, + response_ready_ms = field::Empty, + ) + )] pub(crate) async fn exec_command( &self, request: ExecCommandRequest, context: &UnifiedExecContext, ) -> Result { + let exec_command_started_at = Instant::now(); let cwd = request.cwd.clone(); let process = self .open_session_with_sandbox(&request, cwd.clone(), context) @@ -497,6 +538,10 @@ impl UnifiedExecProcessManager { deadline, ) .await; + Span::current().record( + "output_collected_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); let wall_time = Instant::now().saturating_duration_since(start); let text = String::from_utf8_lossy(&collected).to_string(); @@ -581,6 +626,10 @@ impl UnifiedExecProcessManager { deferred_network_approval.take(), ) .await; + Span::current().record( + "network_approval_complete_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); if let Err(message) = finish_result { emit_failed_initial_exec_end_if_unstored( process_started_alive, @@ -611,25 +660,54 @@ impl UnifiedExecProcessManager { wall_time, ) .await; + Span::current().record( + "end_event_complete_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); - self.release_process_id(request.process_id).await; - process.check_for_sandbox_denial_with_text(&text).await?; + self.release_process_id(request.process_id) + .instrument(tracing::info_span!( + "unified_exec.post_exit.release_process_id", + process_id = request.process_id, + )) + .await; + Span::current().record( + "process_released_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); + process + .check_for_sandbox_denial_with_text(&text) + .instrument(tracing::info_span!( + "unified_exec.post_exit.check_sandbox_denial", + process_id = request.process_id, + )) + .await?; + Span::current().record( + "sandbox_check_complete_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); (None, exit_code) }; - let original_token_count = approx_token_count(&text); - let response = ExecCommandToolOutput { - event_call_id: context.call_id.clone(), - chunk_id, - wall_time, - raw_output: collected, - truncation_policy: context.turn.model_info.truncation_policy.into(), - max_output_tokens: request.max_output_tokens, - process_id: response_process_id, - exit_code, - original_token_count: Some(original_token_count), - hook_command: Some(request.hook_command.clone()), - }; + let response = tracing::info_span!("unified_exec.build_tool_response").in_scope(|| { + let original_token_count = approx_token_count(&text); + ExecCommandToolOutput { + event_call_id: context.call_id.clone(), + chunk_id, + wall_time, + raw_output: collected, + truncation_policy: context.turn.model_info.truncation_policy.into(), + max_output_tokens: request.max_output_tokens, + process_id: response_process_id, + exit_code, + original_token_count: Some(original_token_count), + hook_command: Some(request.hook_command.clone()), + } + }); + Span::current().record( + "response_ready_ms", + exec_command_started_at.elapsed().as_secs_f64() * 1_000.0, + ); Ok(response) } @@ -948,6 +1026,12 @@ impl UnifiedExecProcessManager { }) } + #[instrument( + name = "unified_exec.open_prepared_session", + level = "info", + skip_all, + fields(process_id, remote = environment.is_remote(), tty) + )] pub(crate) async fn open_session_with_prepared_exec_env( &self, process_id: i32, @@ -1051,9 +1135,15 @@ impl UnifiedExecProcessManager { )); } + let params = tracing::info_span!("unified_exec.prepare_exec_server_request") + .in_scope(|| exec_server_params_for_request(process_id, request, tty)); let started = environment .get_exec_backend() - .start(exec_server_params_for_request(process_id, request, tty)) + .start(params) + .instrument(tracing::info_span!( + "unified_exec.remote_backend_start", + process_id, + )) .await .map_err(|err| UnifiedExecError::create_process(err.to_string()))?; spawn_lifecycle.after_spawn(); @@ -1100,16 +1190,30 @@ impl UnifiedExecProcessManager { UnifiedExecProcess::from_spawned(spawned, request.sandbox, spawn_lifecycle).await } + #[instrument( + name = "unified_exec.open_session", + level = "info", + skip_all, + fields( + process_id = request.process_id, + environment_id = %request.turn_environment.environment_id, + remote = request.turn_environment.environment.is_remote(), + tty = request.tty, + ) + )] pub(super) async fn open_session_with_sandbox( &self, request: &ExecCommandRequest, cwd: PathUri, context: &UnifiedExecContext, ) -> Result<(UnifiedExecProcess, Option), UnifiedExecError> { - let local_policy_env = create_env( - &context.turn.config.permissions.shell_environment_policy, - /*thread_id*/ None, - ); + let local_policy_env = tracing::info_span!("unified_exec.prepare_shell_environment") + .in_scope(|| { + create_env( + &context.turn.config.permissions.shell_environment_policy, + /*thread_id*/ None, + ) + }); let mut env = local_policy_env.clone(); env.insert( CODEX_THREAD_ID_ENV_VAR.to_string(), @@ -1142,6 +1246,9 @@ impl UnifiedExecProcessManager { }, prefix_rule: request.prefix_rule.clone(), }) + .instrument(tracing::info_span!( + "unified_exec.resolve_exec_approval_requirement" + )) .await; let req = UnifiedExecToolRequest { command: request.command.clone(), @@ -1183,6 +1290,7 @@ impl UnifiedExecProcessManager { &context.turn, context.turn.approval_policy.value(), ) + .instrument(tracing::info_span!("unified_exec.run_tool_orchestrator")) .await .map(|result| (result.output, result.deferred_network_approval)) .map_err(|err| match err { @@ -1200,6 +1308,16 @@ impl UnifiedExecProcessManager { }) } + #[instrument( + name = "unified_exec.collect_output_until_deadline", + level = "info", + skip_all, + fields( + completion_reason = field::Empty, + exit_signal_received = field::Empty, + collected_bytes = field::Empty, + ) + )] pub(super) async fn collect_output_until_deadline( output_buffer: &OutputBuffer, output_notify: &Arc, @@ -1214,7 +1332,7 @@ impl UnifiedExecProcessManager { let mut collected: Vec = Vec::with_capacity(4096); let mut exit_signal_received = cancellation_token.is_cancelled(); let mut post_exit_deadline: Option = None; - loop { + let completion_reason = loop { Self::extend_deadlines_while_paused( &mut pause_state, &mut deadline, @@ -1235,11 +1353,11 @@ impl UnifiedExecProcessManager { exit_signal_received |= cancellation_token.is_cancelled(); if exit_signal_received && output_closed.load(std::sync::atomic::Ordering::Acquire) { - break; + break "output_closed"; } let remaining = deadline.saturating_duration_since(Instant::now()); if remaining == Duration::ZERO { - break; + break "yield_deadline"; } if exit_signal_received { @@ -1248,7 +1366,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 "post_exit_close_wait_cap"; } let notified = wait_for_output.unwrap_or_else(|| output_notify.notified()); let closed = output_closed_notify.notified(); @@ -1257,7 +1375,9 @@ impl UnifiedExecProcessManager { tokio::select! { _ = &mut notified => {} _ = &mut closed => {} - _ = tokio::time::sleep(close_wait_remaining) => break, + _ = tokio::time::sleep(close_wait_remaining) => { + break "post_exit_close_wait_cap"; + }, _ = Self::wait_for_pause_change(pause_state.as_ref()) => {} } continue; @@ -1270,7 +1390,7 @@ impl UnifiedExecProcessManager { tokio::select! { _ = &mut notified => {} _ = &mut exit_notified => exit_signal_received = true, - _ = tokio::time::sleep(remaining) => break, + _ = tokio::time::sleep(remaining) => break "yield_deadline", _ = Self::wait_for_pause_change(pause_state.as_ref()) => {} } continue; @@ -1282,9 +1402,14 @@ impl UnifiedExecProcessManager { exit_signal_received |= cancellation_token.is_cancelled(); if Instant::now() >= deadline { - break; + break "yield_deadline_after_output"; } - } + }; + + let span = Span::current(); + span.record("completion_reason", completion_reason); + span.record("exit_signal_received", exit_signal_received); + span.record("collected_bytes", collected.len()); collected } diff --git a/codex-rs/core/src/unified_exec/process_tests.rs b/codex-rs/core/src/unified_exec/process_tests.rs index af0cb62741..207bc12cdf 100644 --- a/codex-rs/core/src/unified_exec/process_tests.rs +++ b/codex-rs/core/src/unified_exec/process_tests.rs @@ -172,3 +172,34 @@ async fn remote_terminate_confirmed_updates_state_on_success_only() { assert!(process.has_exited()); } + +#[tokio::test] +async fn remote_process_attach_surfaces_early_sandbox_denial() { + let (wake_tx, _wake_rx) = watch::channel(0); + let started = StartedExecProcess { + process: Arc::new(MockExecProcess { + process_id: "test-process".to_string().into(), + write_response: WriteResponse { + status: WriteStatus::Accepted, + }, + read_responses: Mutex::new(VecDeque::new()), + terminate_error: None, + wake_tx, + }), + }; + + let error = UnifiedExecProcess::from_exec_server_started_with_output_task( + started, + |_started, _output_handles, _output_tx, state_tx| { + tokio::spawn(async move { + let mut state = state_tx.borrow().clone(); + state.sandbox_denied = true; + let _ = state_tx.send_replace(state.exited(Some(1))); + }) + }, + ) + .await + .expect_err("early remote sandbox denial should fail the orchestrated launch attempt"); + + assert!(matches!(error, UnifiedExecError::SandboxDenied { .. })); +}