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

exec-server: trace detached RPC work

Browse Source
This commit is contained in:
Adam Perry
2026-07-09 04:23:49 +00:00
parent 385d7b0016
commit e62c19c428
5 changed files with 152 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
codex-rs/exec-server/src/client/reqwest_http_client.rs
View File

@@ -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(
&notifications,
HttpRequestBodyDeltaNotification {
@@ -255,7 +258,7 @@ impl ReqwestHttpRequestRunner {
}
}
let _ = send_body_delta(
let sent = send_body_delta(
&notifications,
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<HttpHeader>) -> Result<HeaderMap, JSONRPCErrorError> {

26
codex-rs/exec-server/src/file_read.rs
View File

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

75
codex-rs/exec-server/src/local_file_system.rs
View File

@@ -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}")))?

93
codex-rs/exec-server/src/local_process.rs
View File

@@ -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<Notify>,
) {
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) {

27
codex-rs/exec-server/src/server/handler.rs
View File

@@ -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) {