openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Trace detached app-server workflows

Browse Source
This commit is contained in:
Adam Perry
2026-07-10 00:30:40 +00:00
parent bba2192552
commit 4723a8d943
7 changed files with 160 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/app_server_tracing.rs
+4
View File
@@ -32,6 +32,10 @@ where
tokio::task::spawn_blocking(move || span.in_scope(function))
}
pub(crate) fn link_to_current_span(span: &Span) {
span.follows_from(Span::current());
}
pub(crate) fn request_span(
request: &JSONRPCRequest,
transport: &AppServerTransport,
codex-rs/app-server/src/message_processor_tracing_tests.rs
+49
View File
@@ -822,3 +822,52 @@ async fn server_request_span_records_terminal_outcomes() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "current_thread")]
#[serial(app_server_tracing)]
async fn detached_span_is_linked_without_holding_its_cause_open() {
let tracing = init_test_tracing();
tracing.exporter.reset();
let cause = tracing::info_span!("test.app_server.request");
let detached = cause.in_scope(|| {
let span = tracing::info_span!(parent: None, "test.app_server.detached");
crate::app_server_tracing::link_to_current_span(&span);
span
});
drop(cause);
let spans = wait_for_exported_spans(tracing, |spans| {
spans
.iter()
.any(|span| span.name == "test.app_server.request")
})
.await;
assert!(
!spans
.iter()
.any(|span| span.name == "test.app_server.detached")
);
drop(detached);
let spans = wait_for_exported_spans(tracing, |spans| {
spans
.iter()
.any(|span| span.name == "test.app_server.detached")
})
.await;
let cause = spans
.iter()
.find(|span| span.name == "test.app_server.request")
.expect("cause span should export");
let detached = spans
.iter()
.find(|span| span.name == "test.app_server.detached")
.expect("detached span should export");
assert_eq!(detached.parent_span_id, SpanId::INVALID);
assert!(
detached
.links
.iter()
.any(|link| link.span_context.span_id() == cause.span_context.span_id())
);
}
codex-rs/app-server/src/request_processors/account_processor.rs
+32 -6
View File
@@ -235,14 +235,20 @@ impl AccountRequestProcessor {
thread_manager: Arc<ThreadManager>,
config_manager: ConfigManager,
) {
tokio::spawn(async move {
let span = tracing::info_span!(
parent: None,
"app_server.plugins.refresh_after_account_change",
);
crate::app_server_tracing::link_to_current_span(&span);
let refresh_task = async move {
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_service().clear_cache();
if thread_manager.list_thread_ids().await.is_empty() {
return;
}
crate::mcp_refresh::queue_best_effort_refresh(&thread_manager, &config_manager).await;
});
};
tokio::spawn(refresh_task.instrument(span));
}
async fn login_v2(
@@ -464,7 +470,15 @@ impl AccountRequestProcessor {
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let active_login = self.active_login.clone();
let auth_url = server.auth_url.clone();
tokio::spawn(async move {
let login_span = tracing::info_span!(
parent: None,
"app_server.account_login",
account.login.id = %login_id,
account.login.mode = "browser",
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&login_span);
let login_task = async move {
let (success, error_msg) = match tokio::time::timeout(
LOGIN_CHATGPT_TIMEOUT,
server.block_until_done(),
@@ -478,6 +492,7 @@ impl AccountRequestProcessor {
(false, Some("Login timed out".to_string()))
}
};
tracing::Span::current().record("result", if success { "success" } else { "error" });
Self::send_chatgpt_login_completion_notifications(
&outgoing_clone,
@@ -495,7 +510,8 @@ impl AccountRequestProcessor {
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
};
tokio::spawn(login_task.instrument(login_span));
Ok(LoginAccountResponse::Chatgpt {
login_id: login_id.to_string(),
@@ -542,7 +558,15 @@ impl AccountRequestProcessor {
let thread_manager = Arc::clone(&self.thread_manager);
let chatgpt_base_url = self.config.chatgpt_base_url.clone();
let active_login = self.active_login.clone();
tokio::spawn(async move {
let login_span = tracing::info_span!(
parent: None,
"app_server.account_login",
account.login.id = %login_id,
account.login.mode = "device_code",
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&login_span);
let login_task = async move {
let (success, error_msg) = tokio::select! {
_ = cancel.cancelled() => {
(false, Some("Login was not completed".to_string()))
@@ -554,6 +578,7 @@ impl AccountRequestProcessor {
}
}
};
tracing::Span::current().record("result", if success { "success" } else { "error" });
Self::send_chatgpt_login_completion_notifications(
&outgoing_clone,
@@ -570,7 +595,8 @@ impl AccountRequestProcessor {
if guard.as_ref().map(ActiveLogin::login_id) == Some(login_id) {
*guard = None;
}
});
};
tokio::spawn(login_task.instrument(login_span));
Ok(LoginAccountResponse::ChatgptDeviceCode {
login_id: login_id.to_string(),
codex-rs/app-server/src/request_processors/external_agent_config_processor.rs
+22 -2
View File
@@ -48,6 +48,7 @@ use codex_state::ExternalAgentConfigImportSuccessRecord;
use codex_thread_store::ThreadStore;
use std::collections::HashSet;
use std::path::PathBuf;
use tracing::Instrument;
use super::ConfigRequestProcessor;
use super::external_agent_session_import::ExternalAgentSessionImporter;
@@ -281,7 +282,16 @@ impl ExternalAgentConfigRequestProcessor {
)
});
let pending_plugin_imports = import_outcome.pending_plugin_imports;
tokio::spawn(async move {
let background_import_span = tracing::info_span!(
parent: None,
"app_server.external_agent_config_import",
external_agent_config.import.id = %import_id,
external_agent_config.import.session_count = pending_session_imports.len(),
external_agent_config.import.plugin_count = pending_plugin_imports.len(),
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&background_import_span);
let background_import_task = async move {
let session_progress_outgoing = Arc::clone(&outgoing);
let session_import_id = import_id.clone();
let session_imports = async move {
@@ -340,6 +350,15 @@ impl ExternalAgentConfigRequestProcessor {
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_service().clear_cache();
}
let result = if completed_item_results
.iter()
.any(|item_result| item_result.error_count > 0)
{
"partial_error"
} else {
"success"
};
tracing::Span::current().record("result", result);
send_completed_import_notification(
&outgoing,
state_db.as_ref(),
@@ -349,7 +368,8 @@ impl ExternalAgentConfigRequestProcessor {
&completed_item_results,
)
.await;
});
};
tokio::spawn(background_import_task.instrument(background_import_span));
Ok(())
}
codex-rs/app-server/src/request_processors/mcp_processor.rs
+11 -2
View File
@@ -201,12 +201,20 @@ impl McpRequestProcessor {
let notification_name = name.clone();
let notification_thread_id = thread_id;
let outgoing = Arc::clone(&self.outgoing);
let oauth_span = tracing::info_span!(
parent: None,
"app_server.mcp_oauth_login",
codex.thread.id = notification_thread_id.as_deref().unwrap_or(""),
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&oauth_span);
tokio::spawn(async move {
let oauth_task = async move {
let (success, error) = match handle.wait().await {
Ok(()) => (true, None),
Err(err) => (false, Some(err.to_string())),
};
tracing::Span::current().record("result", if success { "success" } else { "error" });
let notification = ServerNotification::McpServerOauthLoginCompleted(
McpServerOauthLoginCompletedNotification {
@@ -217,7 +225,8 @@ impl McpRequestProcessor {
},
);
outgoing.send_server_notification(notification).await;
});
};
tokio::spawn(oauth_task.instrument(oauth_span));
Ok(McpServerOauthLoginResponse { authorization_url })
}
codex-rs/app-server/src/request_processors/plugins.rs
+19 -4
View File
@@ -490,14 +490,20 @@ impl PluginRequestProcessor {
thread_manager: Arc<ThreadManager>,
config_manager: ConfigManager,
) {
tokio::spawn(async move {
let span = tracing::info_span!(
parent: None,
"app_server.plugins.refresh_after_change",
);
crate::app_server_tracing::link_to_current_span(&span);
let refresh_task = async move {
thread_manager.plugins_manager().clear_cache();
thread_manager.skills_service().clear_cache();
if thread_manager.list_thread_ids().await.is_empty() {
return;
}
crate::mcp_refresh::queue_best_effort_refresh(&thread_manager, &config_manager).await;
});
};
tokio::spawn(refresh_task.instrument(span));
}
fn clear_plugin_related_caches(&self) {
@@ -1862,8 +1868,14 @@ impl PluginRequestProcessor {
let callback_url = config.mcp_oauth_callback_url.clone();
let outgoing = Arc::clone(&self.outgoing);
let notification_name = name.clone();
let oauth_span = tracing::info_span!(
parent: None,
"app_server.plugin_mcp_oauth_login",
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&oauth_span);
tokio::spawn(async move {
let oauth_task = async move {
let oauth_client_id = server.oauth_client_id();
let first_attempt = perform_oauth_login_silent(
&name,
@@ -1904,6 +1916,8 @@ impl PluginRequestProcessor {
Ok(()) => (true, None),
Err(err) => (false, Some(err.to_string())),
};
tracing::Span::current()
.record("result", if success { "success" } else { "error" });
let notification = ServerNotification::McpServerOauthLoginCompleted(
McpServerOauthLoginCompletedNotification {
@@ -1914,7 +1928,8 @@ impl PluginRequestProcessor {
},
);
outgoing.send_server_notification(notification).await;
});
};
tokio::spawn(oauth_task.instrument(oauth_span));
}
}
codex-rs/app-server/src/request_processors/windows_sandbox_processor.rs
+23 -2
View File
@@ -73,8 +73,20 @@ impl WindowsSandboxRequestProcessor {
let outgoing = Arc::clone(&self.outgoing);
let connection_id = request_id.connection_id;
let setup_mode_name = match setup_mode {
CoreWindowsSandboxSetupMode::Elevated => "elevated",
CoreWindowsSandboxSetupMode::Unelevated => "unelevated",
};
let setup_span = tracing::info_span!(
parent: None,
"app_server.windows_sandbox_setup",
app_server.connection_id = %connection_id,
windows_sandbox.setup.mode = setup_mode_name,
result = tracing::field::Empty,
);
crate::app_server_tracing::link_to_current_span(&setup_span);
tokio::spawn(async move {
let setup_task = async move {
let setup_request = WindowsSandboxSetupRequest {
mode: setup_mode,
permission_profile: config.permissions.effective_permission_profile(),
@@ -85,6 +97,14 @@ impl WindowsSandboxRequestProcessor {
};
let setup_result =
codex_core::windows_sandbox::run_windows_sandbox_setup(setup_request).await;
tracing::Span::current().record(
"result",
if setup_result.is_ok() {
"success"
} else {
"error"
},
);
let notification = WindowsSandboxSetupCompletedNotification {
mode: match setup_mode {
CoreWindowsSandboxSetupMode::Elevated => WindowsSandboxSetupMode::Elevated,
@@ -99,7 +119,8 @@ impl WindowsSandboxRequestProcessor {
ServerNotification::WindowsSandboxSetupCompleted(notification),
)
.await;
});
};
tokio::spawn(setup_task.instrument(setup_span));
Ok(())
}
}