diff --git a/codex-rs/exec-server/src/client/reqwest_http_client.rs b/codex-rs/exec-server/src/client/reqwest_http_client.rs index d0cffa6a32..f8033f4046 100644 --- a/codex-rs/exec-server/src/client/reqwest_http_client.rs +++ b/codex-rs/exec-server/src/client/reqwest_http_client.rs @@ -234,11 +234,14 @@ impl ReqwestHttpRequestRunner { ) .await { + tracing::Span::current().record("result", "disconnected"); return; } seq += 1; } Err(error) => { + tracing::Span::current().record("result", "error"); + tracing::Span::current().record("error.type", "response_body"); let _ = send_body_delta( ¬ifications, HttpRequestBodyDeltaNotification { @@ -255,7 +258,7 @@ impl ReqwestHttpRequestRunner { } } - let _ = send_body_delta( + let sent = send_body_delta( ¬ifications, HttpRequestBodyDeltaNotification { request_id, @@ -266,6 +269,7 @@ impl ReqwestHttpRequestRunner { }, ) .await; + tracing::Span::current().record("result", if sent { "success" } else { "disconnected" }); } fn build_headers(headers: Vec) -> Result { diff --git a/codex-rs/exec-server/src/file_read.rs b/codex-rs/exec-server/src/file_read.rs index 8cc84b6730..b716c4dba8 100644 --- a/codex-rs/exec-server/src/file_read.rs +++ b/codex-rs/exec-server/src/file_read.rs @@ -57,13 +57,25 @@ impl FileReadHandleManager { .cloned() .ok_or_else(|| unknown_handle_error(handle_id))? }; - let result = - match tokio::task::spawn_blocking(move || read_block_at(&file, offset, len)).await { - Ok(result) => result, - Err(error) => Err(io::Error::other(format!( - "file read task stopped unexpectedly: {error}" - ))), - }; + let read_span = tracing::info_span!( + parent: None, + "codex.exec_server.fs_read_block", + otel.kind = "internal", + fs.handle_id = handle_id, + fs.offset = offset, + fs.length = len, + ); + read_span.follows_from(tracing::Span::current()); + let result = match tokio::task::spawn_blocking(move || { + read_span.in_scope(|| read_block_at(&file, offset, len)) + }) + .await + { + Ok(result) => result, + Err(error) => Err(io::Error::other(format!( + "file read task stopped unexpectedly: {error}" + ))), + }; if result.is_err() { self.close(handle_id).await; } diff --git a/codex-rs/exec-server/src/local_file_system.rs b/codex-rs/exec-server/src/local_file_system.rs index 281fd1eeb3..76a305842a 100644 --- a/codex-rs/exec-server/src/local_file_system.rs +++ b/codex-rs/exec-server/src/local_file_system.rs @@ -667,44 +667,53 @@ impl DirectFileSystem { reject_sandbox_context(sandbox)?; let source_path = source_path.to_abs_path()?.into_path_buf(); let destination_path = destination_path.to_abs_path()?.into_path_buf(); - tokio::task::spawn_blocking(move || -> FileSystemResult<()> { - let metadata = std::fs::symlink_metadata(source_path.as_path())?; - let file_type = metadata.file_type(); + let copy_span = tracing::info_span!( + parent: None, + "codex.exec_server.fs_copy", + otel.kind = "internal", + fs.recursive = options.recursive, + ); + copy_span.follows_from(tracing::Span::current()); + tokio::task::spawn_blocking(move || { + copy_span.in_scope(|| -> FileSystemResult<()> { + let metadata = std::fs::symlink_metadata(source_path.as_path())?; + let file_type = metadata.file_type(); - if file_type.is_dir() { - if !options.recursive { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "fs/copy requires recursive: true when sourcePath is a directory", - )); + if file_type.is_dir() { + if !options.recursive { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "fs/copy requires recursive: true when sourcePath is a directory", + )); + } + if destination_is_same_or_descendant_of_source( + source_path.as_path(), + destination_path.as_path(), + )? { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "fs/copy cannot copy a directory to itself or one of its descendants", + )); + } + copy_dir_recursive(source_path.as_path(), destination_path.as_path())?; + return Ok(()); } - if destination_is_same_or_descendant_of_source( - source_path.as_path(), - destination_path.as_path(), - )? { - return Err(io::Error::new( - io::ErrorKind::InvalidInput, - "fs/copy cannot copy a directory to itself or one of its descendants", - )); + + if file_type.is_symlink() { + copy_symlink(source_path.as_path(), destination_path.as_path())?; + return Ok(()); } - copy_dir_recursive(source_path.as_path(), destination_path.as_path())?; - return Ok(()); - } - if file_type.is_symlink() { - copy_symlink(source_path.as_path(), destination_path.as_path())?; - return Ok(()); - } + if file_type.is_file() { + std::fs::copy(source_path.as_path(), destination_path.as_path())?; + return Ok(()); + } - if file_type.is_file() { - std::fs::copy(source_path.as_path(), destination_path.as_path())?; - return Ok(()); - } - - Err(io::Error::new( - io::ErrorKind::InvalidInput, - "fs/copy only supports regular files, directories, and symlinks", - )) + Err(io::Error::new( + io::ErrorKind::InvalidInput, + "fs/copy only supports regular files, directories, and symlinks", + )) + }) }) .await .map_err(|err| io::Error::other(format!("filesystem task failed: {err}")))? diff --git a/codex-rs/exec-server/src/local_process.rs b/codex-rs/exec-server/src/local_process.rs index 3c488f0ad7..3199358141 100644 --- a/codex-rs/exec-server/src/local_process.rs +++ b/codex-rs/exec-server/src/local_process.rs @@ -22,6 +22,7 @@ use tokio::sync::Mutex; use tokio::sync::Notify; use tokio::sync::mpsc; use tokio::sync::watch; +use tracing::Instrument; use crate::ExecBackend; use crate::ExecBackendFuture; @@ -341,34 +342,69 @@ impl LocalProcess { })), ); } - tokio::spawn(stream_output( - process_id.clone(), - if params.tty { - ExecOutputStream::Pty - } else { - ExecOutputStream::Stdout - }, - spawned.stdout_rx, - Arc::clone(&self.inner), - Arc::clone(&output_notify), - )); - tokio::spawn(stream_output( - process_id.clone(), - if params.tty { - ExecOutputStream::Pty - } else { - ExecOutputStream::Stderr - }, - spawned.stderr_rx, - Arc::clone(&self.inner), - Arc::clone(&output_notify), - )); - tokio::spawn(watch_exit( - process_id.clone(), - spawned.exit_rx, - Arc::clone(&self.inner), - output_notify, - )); + let (stdout_stream, stdout_stream_name) = if params.tty { + (ExecOutputStream::Pty, "pty") + } else { + (ExecOutputStream::Stdout, "stdout") + }; + let stdout_span = tracing::info_span!( + parent: None, + "codex.exec_server.process_output", + otel.kind = "internal", + exec_server.process_id = %process_id, + exec_server.output_stream = stdout_stream_name, + ); + stdout_span.follows_from(tracing::Span::current()); + tokio::spawn( + stream_output( + process_id.clone(), + stdout_stream, + spawned.stdout_rx, + Arc::clone(&self.inner), + Arc::clone(&output_notify), + ) + .instrument(stdout_span), + ); + let (stderr_stream, stderr_stream_name) = if params.tty { + (ExecOutputStream::Pty, "pty") + } else { + (ExecOutputStream::Stderr, "stderr") + }; + let stderr_span = tracing::info_span!( + parent: None, + "codex.exec_server.process_output", + otel.kind = "internal", + exec_server.process_id = %process_id, + exec_server.output_stream = stderr_stream_name, + ); + stderr_span.follows_from(tracing::Span::current()); + tokio::spawn( + stream_output( + process_id.clone(), + stderr_stream, + spawned.stderr_rx, + Arc::clone(&self.inner), + Arc::clone(&output_notify), + ) + .instrument(stderr_span), + ); + let wait_span = tracing::info_span!( + parent: None, + "codex.exec_server.process_wait", + otel.kind = "internal", + exec_server.process_id = %process_id, + process.exit_code = tracing::field::Empty, + ); + wait_span.follows_from(tracing::Span::current()); + tokio::spawn( + watch_exit( + process_id.clone(), + spawned.exit_rx, + Arc::clone(&self.inner), + output_notify, + ) + .instrument(wait_span), + ); Ok((ExecResponse { process_id }, wake_tx, events)) } @@ -823,6 +859,7 @@ async fn watch_exit( output_notify: Arc, ) { let exit_code = exit_rx.await.unwrap_or(-1); + tracing::Span::current().record("process.exit_code", exit_code); let sandboxed = { let mut processes = inner.processes.lock().await; match processes.get_mut(&process_id) { diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 02858ae887..c4a7a37caa 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -10,6 +10,7 @@ use std::collections::HashSet; use tokio::sync::Mutex; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; +use tracing::Instrument; use crate::ExecServerRuntimePaths; use crate::client::http_client::PendingReqwestHttpBodyStream; @@ -390,13 +391,27 @@ impl ExecServerHandler { let handler = Arc::clone(self); let notifications = self.notifications.clone(); let shutdown = self.background_task_shutdown.clone(); - self.background_tasks.spawn(async move { - tokio::select! { - _ = shutdown.cancelled() => {} - _ = ReqwestHttpRequestRunner::stream_body(pending_stream, notifications) => {} + let stream_span = tracing::info_span!( + parent: None, + "codex.exec_server.http_response_body", + otel.kind = "internal", + exec_server.http_request_id = request_id, + result = tracing::field::Empty, + error.type = tracing::field::Empty, + ); + stream_span.follows_from(tracing::Span::current()); + self.background_tasks.spawn( + async move { + tokio::select! { + _ = shutdown.cancelled() => { + tracing::Span::current().record("result", "cancelled"); + } + _ = ReqwestHttpRequestRunner::stream_body(pending_stream, notifications) => {} + } + handler.release_http_body_stream(&finished_request_id).await; } - handler.release_http_body_stream(&finished_request_id).await; - }); + .instrument(stream_span), + ); } async fn release_http_body_stream(&self, request_id: &str) {