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

telemetry(core): trace unified exec process lifecycle

Browse Source
This commit is contained in:
Anton Panasenko
2026-06-29 22:09:35 -07:00
parent 6afcf26d5d
commit eb06bdcc87
4 changed files with 486 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
codex-rs/core/src/unified_exec/async_watcher.rs
View File

@@ -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<Session>,
turn_ref: Arc<TurnContext>,
@@ -317,6 +319,11 @@ fn split_valid_utf8_prefix_with_max(buffer: &mut Vec<u8>, max_bytes: usize) -> O
Some(byte)
}
#[instrument(
name = "unified_exec.resolve_aggregated_output",
level = "info",
skip_all
)]
async fn resolve_aggregated_output(
transcript: &Arc<Mutex<HeadTailBuffer>>,
fallback: String,

432
codex-rs/core/src/unified_exec/process.rs
View File

@@ -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<u8> = 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, UnifiedExecError> {
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<Vec<u8>>,
watch::Sender<ProcessState>,
) -> JoinHandle<()>,
) -> Result<Self, UnifiedExecError> {
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(

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

@@ -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<CancellationToken>) -> 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<crate::session::session::Session>>,
deferred: Option<DeferredNetworkApproval>,
@@ -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<ExecCommandToolOutput, UnifiedExecError> {
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<DeferredNetworkApproval>), 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<Notify>,
@@ -1214,7 +1332,7 @@ impl UnifiedExecProcessManager {
let mut collected: Vec<u8> = Vec::with_capacity(4096);
let mut exit_signal_received = cancellation_token.is_cancelled();
let mut post_exit_deadline: Option<Instant> = 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
}

31
codex-rs/core/src/unified_exec/process_tests.rs
View File

@@ -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 { .. }));
}