From 4723a8d9432a05d858cbaa877728e8e9f59c6936 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 9 Jul 2026 04:15:36 +0000 Subject: [PATCH] Trace detached app-server workflows --- codex-rs/app-server/src/app_server_tracing.rs | 4 ++ .../src/message_processor_tracing_tests.rs | 49 +++++++++++++++++++ .../request_processors/account_processor.rs | 38 +++++++++++--- .../external_agent_config_processor.rs | 24 ++++++++- .../src/request_processors/mcp_processor.rs | 13 ++++- .../src/request_processors/plugins.rs | 23 +++++++-- .../windows_sandbox_processor.rs | 25 +++++++++- 7 files changed, 160 insertions(+), 16 deletions(-) diff --git a/codex-rs/app-server/src/app_server_tracing.rs b/codex-rs/app-server/src/app_server_tracing.rs index 7ff9c24b00..e50cbd587c 100644 --- a/codex-rs/app-server/src/app_server_tracing.rs +++ b/codex-rs/app-server/src/app_server_tracing.rs @@ -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, diff --git a/codex-rs/app-server/src/message_processor_tracing_tests.rs b/codex-rs/app-server/src/message_processor_tracing_tests.rs index 6f755e6c38..1829a0dc1e 100644 --- a/codex-rs/app-server/src/message_processor_tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor_tracing_tests.rs @@ -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()) + ); +} diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index d98977ce58..4c71fee5b5 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -235,14 +235,20 @@ impl AccountRequestProcessor { thread_manager: Arc, 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(), diff --git a/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs b/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs index c71abcb884..60d699cb20 100644 --- a/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs +++ b/codex-rs/app-server/src/request_processors/external_agent_config_processor.rs @@ -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(()) } diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index 60041e39c0..e03b39a39e 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -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 }) } diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 8d672eaab5..a361caab04 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -490,14 +490,20 @@ impl PluginRequestProcessor { thread_manager: Arc, 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)); } } diff --git a/codex-rs/app-server/src/request_processors/windows_sandbox_processor.rs b/codex-rs/app-server/src/request_processors/windows_sandbox_processor.rs index c0917d933b..d106fb60a1 100644 --- a/codex-rs/app-server/src/request_processors/windows_sandbox_processor.rs +++ b/codex-rs/app-server/src/request_processors/windows_sandbox_processor.rs @@ -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(()) } }