From 60825b498810dd652405ec6f2a1994ee3a73fa8d Mon Sep 17 00:00:00 2001 From: priyanshusingh-de Date: Thu, 10 Sep 2026 21:20:44 +0000 Subject: [PATCH] Honor thread analytics opt-outs when using shared clients (#44646) ## Why An enabled shared analytics client could override a thread's explicit opt-out. Delegated threads also emitted initialization events through the parent's client, bypassing the child's analytics setting. ## What changed - Use a disabled analytics client when `config.analytics_enabled` is `Some(false)`, without disabling analytics for sibling threads or overriding a disabled host client. - Emit delegated thread initialization events through the child's analytics client. - Expose the effective analytics state through `CodexThread::analytics_enabled()`. ## Testing Add regression tests for explicit and unset thread settings with enabled, disabled, and absent shared clients, plus delegated child opt-outs with an enabled parent. Update compaction and rollout-budget rollback tests to wait for thread idle after turn completion and fail immediately on rollback errors. GitOrigin-RevId: 547852909af1a6822b95c0bcd2336b0d6bc7aad1 --- codex-rs/core/src/codex_delegate.rs | 2 +- codex-rs/core/src/codex_delegate_tests.rs | 80 +++++++++++ codex-rs/core/src/codex_thread.rs | 5 + codex-rs/core/src/session/session.rs | 18 ++- codex-rs/core/src/thread_manager_tests.rs | 127 ++++++++++++++++++ codex-rs/core/tests/common/lib.rs | 34 +++++ .../core/tests/suite/compact_resume_fork.rs | 38 ++++-- codex-rs/core/tests/suite/rollout_budget.rs | 10 ++ 8 files changed, 295 insertions(+), 19 deletions(-) diff --git a/codex-rs/core/src/codex_delegate.rs b/codex-rs/core/src/codex_delegate.rs index d5795df0ff..4a3e9c4a5d 100644 --- a/codex-rs/core/src/codex_delegate.rs +++ b/codex-rs/core/src/codex_delegate.rs @@ -145,7 +145,7 @@ pub(crate) async fn run_codex_thread_interactive( let thread_config = session.thread_config_snapshot().await; let client_metadata = parent_session.app_server_client_metadata().await; emit_subagent_session_started( - &parent_session.services.analytics_events_client, + &session.services.analytics_events_client, client_metadata, session.session_id(), session.thread_id(), diff --git a/codex-rs/core/src/codex_delegate_tests.rs b/codex-rs/core/src/codex_delegate_tests.rs index 5c31a2531b..d2b5f31add 100644 --- a/codex-rs/core/src/codex_delegate_tests.rs +++ b/codex-rs/core/src/codex_delegate_tests.rs @@ -217,6 +217,86 @@ async fn run_codex_thread_interactive_respects_pre_cancelled_spawn() { )); } +#[tokio::test] +async fn delegate_start_analytics_honors_child_opt_out_with_enabled_parent() { + use codex_analytics::AnalyticsEventsClient; + use codex_login::CodexAuth; + use wiremock::Mock; + use wiremock::MockServer; + use wiremock::ResponseTemplate; + use wiremock::matchers::path; + + let server = MockServer::start().await; + Mock::given(path("/codex/analytics-events/events")) + .respond_with(ResponseTemplate::new(200)) + .mount(&server) + .await; + let client = AnalyticsEventsClient::new( + AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()), + server.uri(), + /*analytics_enabled*/ Some(true), + ); + let (mut parent_session, parent_ctx, _rx_events) = + crate::session::tests::make_session_and_context_with_rx().await; + Arc::get_mut(&mut parent_session) + .expect("parent session should be uniquely owned") + .services + .analytics_events_client = client.clone(); + parent_session + .set_app_server_client_info( + Some("codex-test".to_string()), + Some("1.0.0".to_string()), + /*mcp_elicitations_auto_deny*/ false, + ) + .await + .expect("set parent client metadata"); + + let mut expected_events = Vec::new(); + for analytics_enabled in [false, true] { + let mut config = parent_ctx.config.as_ref().clone(); + config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never); + config.analytics_enabled = Some(analytics_enabled); + let (session, io) = run_codex_thread_interactive( + config, + Arc::clone(&parent_session.services.auth_manager), + Arc::clone(&parent_session.services.models_manager), + Arc::clone(&parent_session), + Arc::clone(&parent_ctx), + parent_ctx.environments.clone(), + CancellationToken::new(), + SubAgentSource::Review, + codex_extension_api::SessionIsolation::Inherit, + /*initial_history*/ None, + crate::session::GitEnrichmentPolicy::Fresh, + codex_sandboxing::WindowsSandboxProxySettingsMode::Reconcile, + ) + .await + .expect("delegate session should start"); + if analytics_enabled { + expected_events.push(serde_json::json!([ + "codex_thread_initialized", + session.thread_id().to_string(), + ])); + } + io.shutdown_and_wait() + .await + .expect("delegate session should shut down"); + } + client.flush().await; + let events = server + .received_requests() + .await + .expect("analytics requests") + .into_iter() + .flat_map(|request| { + let payload: Value = serde_json::from_slice(&request.body).expect("analytics payload"); + payload["events"].as_array().expect("events array").clone() + }) + .map(|event| serde_json::json!([event["event_type"], event["event_params"]["thread_id"]])) + .collect::>(); + assert_eq!(events, expected_events); +} + #[tokio::test] async fn delegate_isolation_does_not_depend_on_attribution() { let (mut parent_session, parent_ctx, _rx_events) = diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 7177740fea..9fa43519f2 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -232,6 +232,11 @@ impl CodexThread { self.session.services.session_telemetry.clone() } + /// Whether analytics is enabled for this thread after configuration and host overrides. + pub fn analytics_enabled(&self) -> bool { + self.session.services.analytics_events_client.is_enabled() + } + /// Returns extension-owned data attached to this thread runtime. pub fn thread_extension_data(&self) -> &codex_extension_api::ExtensionData { &self.session.services.thread_extension_data diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 3a6cbfd137..d949d77ed7 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -1438,13 +1438,17 @@ impl Session { }); } - let analytics_events_client = analytics_events_client.unwrap_or_else(|| { - AnalyticsEventsClient::new( - Arc::clone(&auth_manager), - config.chatgpt_base_url.trim_end_matches('/').to_string(), - config.analytics_enabled, - ) - }); + let analytics_events_client = if config.analytics_enabled == Some(false) { + AnalyticsEventsClient::disabled() + } else { + analytics_events_client.unwrap_or_else(|| { + AnalyticsEventsClient::new( + Arc::clone(&auth_manager), + config.chatgpt_base_url.trim_end_matches('/').to_string(), + config.analytics_enabled, + ) + }) + }; for item in initial_history.get_rollout_items() { match item { RolloutItem::Compacted(compacted) => { diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index c25a55575c..ec516141b0 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -53,6 +53,133 @@ use wiremock::MockServer; const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111"; +/// A thread opt-out wins over a shared client without disabling its siblings. +#[tokio::test] +async fn thread_analytics_opt_out_overrides_shared_client() { + let server = MockServer::start().await; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/codex/analytics-events/events")) + .respond_with(wiremock::ResponseTemplate::new(200)) + .mount(&server) + .await; + let temp_dir = tempdir().expect("tempdir"); + let mut config = test_config().await; + config.chatgpt_base_url = server.uri(); + config.model_provider.base_url = Some(server.uri()); + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + let shared_client = AnalyticsEventsClient::new( + AuthManager::from_auth_for_testing(auth.clone()), + server.uri(), + /*analytics_enabled*/ Some(true), + ); + let mut expected_thread_ids = Vec::new(); + let mut opted_out_thread_ids = Vec::new(); + + for (name, client_override, expected_enabled) in [ + ( + "enabled_override", + Some(shared_client.clone()), + [false, true, true], + ), + ( + "disabled_override", + Some(AnalyticsEventsClient::disabled()), + [false, false, false], + ), + ("no_override", None, [false, true, true]), + ] { + config.codex_home = temp_dir.path().join(name).abs(); + config.cwd = config.codex_home.abs(); + std::fs::create_dir_all(&config.codex_home).expect("create codex home"); + let mut manager = ThreadManager::with_models_provider_and_home_for_tests( + auth.clone(), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ); + Arc::get_mut(&mut manager.state) + .expect("unshared thread manager state") + .analytics_events_client = client_override; + + for (setting, enabled) in [Some(false), Some(true), None] + .into_iter() + .zip(expected_enabled) + { + config.analytics_enabled = setting; + let started = manager + .start_thread(StartThreadOptions::new(config.clone())) + .await + .expect("start analytics test thread"); + let services = &started.thread.session.services; + assert_eq!(started.thread.analytics_enabled(), enabled); + assert_eq!( + services + .session_extension_data + .get::() + .expect("analytics client in session store") + .is_enabled(), + enabled, + ); + let thread_id = started.thread_id.to_string(); + if enabled { + expected_thread_ids.push(thread_id.clone()); + } else { + opted_out_thread_ids.push(thread_id.clone()); + } + services.analytics_events_client.track_app_used( + codex_analytics::TrackEventsContext { + model_slug: "test-model".to_string(), + turn_id: format!("test-turn-{thread_id}"), + thread_id, + product_client_id: "codex_work_cca".to_string(), + }, + codex_analytics::AppInvocation { + connector_id: Some("test-connector".to_string()), + app_name: None, + invocation_type: None, + }, + ); + services.analytics_events_client.flush().await; + } + let shutdown = manager + .shutdown_all_threads_bounded(Duration::from_secs(10)) + .await; + assert_eq!(shutdown.completed.len(), 3); + } + + let events: Vec = server + .received_requests() + .await + .expect("analytics requests") + .into_iter() + .filter(|request| request.url.path() == "/codex/analytics-events/events") + .flat_map(|request| { + request.body_json::().expect("JSON body")["events"] + .as_array() + .expect("events array") + .clone() + }) + .collect(); + assert!(events.iter().all(|event| { + !opted_out_thread_ids + .iter() + .any(|thread_id| event["event_params"]["thread_id"] == thread_id.as_str()) + })); + let mut actual_thread_ids: Vec = events + .iter() + .filter(|event| event["event_type"] == "codex_app_used") + .map(|event| { + event["event_params"]["thread_id"] + .as_str() + .expect("app usage thread ID") + .to_string() + }) + .collect(); + actual_thread_ids.sort(); + expected_thread_ids.sort(); + assert_eq!(actual_thread_ids, expected_thread_ids); +} + /// Controls without a custom allocation policy still produce distinct thread identifiers. #[test] fn thread_id_generator_defaults_to_standard_ids() { diff --git a/codex-rs/core/tests/common/lib.rs b/codex-rs/core/tests/common/lib.rs index 570535558d..424cdb0f9c 100644 --- a/codex-rs/core/tests/common/lib.rs +++ b/codex-rs/core/tests/common/lib.rs @@ -329,6 +329,40 @@ pub async fn submit_thread_settings( } } +/// For sequential tests, register this contributor and wait once after every completed turn. +/// Notifications are thread-scoped so a child or sibling cannot satisfy the wait. +pub struct ThreadIdle; + +#[derive(Default)] +struct ThreadIdleNotification(tokio::sync::Notify); + +impl codex_extension_api::ThreadLifecycleContributor for ThreadIdle { + fn on_thread_idle<'a>( + &'a self, + input: codex_extension_api::ThreadIdleInput<'a>, + ) -> codex_extension_api::ExtensionFuture<'a, ()> { + Box::pin(async move { + input + .thread_store + .get_or_init(ThreadIdleNotification::default) + .0 + .notify_one(); + }) + } +} + +impl ThreadIdle { + pub async fn wait(thread: &CodexThread) { + // TurnComplete is sent before active-turn cleanup. Rollback requires the later idle signal. + let idle = thread + .thread_extension_data() + .get_or_init(ThreadIdleNotification::default); + tokio::time::timeout(std::time::Duration::from_secs(10), idle.0.notified()) + .await + .expect("thread should become idle after turn completion"); + } +} + pub async fn wait_for_event_match(codex: &CodexThread, matcher: F) -> T where F: Fn(&codex_protocol::protocol::EventMsg) -> Option, diff --git a/codex-rs/core/tests/suite/compact_resume_fork.rs b/codex-rs/core/tests/suite/compact_resume_fork.rs index d937ad13c4..b60a453d0a 100644 --- a/codex-rs/core/tests/suite/compact_resume_fork.rs +++ b/codex-rs/core/tests/suite/compact_resume_fork.rs @@ -16,6 +16,7 @@ use codex_core::TurnInputRequest; use codex_core::compact::SUMMARIZATION_PROMPT; use codex_core::config::Config; use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR; +use codex_extension_api::ExtensionRegistryBuilder; use codex_history::CodexHarnessMetadata; use codex_history::RolloutItem; use codex_protocol::config_types::CollaborationMode; @@ -29,6 +30,7 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::WarningEvent; use codex_protocol::user_input::UserInput; +use core_test_support::ThreadIdle; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::context_snapshot::ContextSnapshotRenderMode; @@ -543,8 +545,13 @@ async fn snapshot_rollback_past_compaction_replays_append_only_history() -> Resu base.submit(Op::ThreadRollback { num_turns: 1 }) .await .expect("submit thread rollback"); - let rollback_event = - wait_for_event(&base, |ev| matches!(ev, EventMsg::ThreadRolledBack(_))).await; + let rollback_event = wait_for_event(&base, |ev| { + if let EventMsg::Error(error) = ev { + panic!("rollback failed: {error:?}"); + } + matches!(ev, EventMsg::ThreadRolledBack(_)) + }) + .await; let EventMsg::ThreadRolledBack(rollback_event) = rollback_event else { panic!("expected thread rolled back event"); }; @@ -658,6 +665,9 @@ async fn snapshot_rollback_followup_turn_trims_context_updates() -> Result<()> { .submit(Op::ThreadRollback { num_turns: 1 }) .await?; let rollback_event = wait_for_event(&conversation, |ev| { + if let EventMsg::Error(error) = ev { + panic!("rollback failed: {error:?}"); + } matches!(ev, EventMsg::ThreadRolledBack(_)) }) .await; @@ -844,15 +854,19 @@ async fn start_test_conversation( ) -> (Arc, Config, Arc, Arc) { let base_url = format!("{}/v1", server.uri()); let model = model.map(str::to_string); - let mut builder = test_codex().with_config(move |config| { - config.update_plan_enabled = true; - config.model_provider.name = "Non-OpenAI Model provider".to_string(); - config.model_provider.base_url = Some(base_url); - config.compact_prompt = Some(SUMMARIZATION_PROMPT.to_string()); - if let Some(model) = model { - config.model = Some(model); - } - }); + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle)); + let mut builder = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_config(move |config| { + config.update_plan_enabled = true; + config.model_provider.name = "Non-OpenAI Model provider".to_string(); + config.model_provider.base_url = Some(base_url); + config.compact_prompt = Some(SUMMARIZATION_PROMPT.to_string()); + if let Some(model) = model { + config.model = Some(model); + } + }); let test = Box::pin(builder.build(server)) .await .expect("create conversation"); @@ -868,6 +882,7 @@ async fn user_turn(conversation: &Arc, text: &str) { .await .expect("submit user turn"); wait_for_event(conversation, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + ThreadIdle::wait(conversation).await; } async fn compact_conversation(conversation: &Arc) { @@ -887,6 +902,7 @@ async fn compact_conversation(conversation: &Arc) { }; assert_eq!(message, COMPACT_WARNING_MESSAGE); wait_for_event(conversation, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + ThreadIdle::wait(conversation).await; } fn fetch_conversation_path(conversation: &Arc) -> std::path::PathBuf { diff --git a/codex-rs/core/tests/suite/rollout_budget.rs b/codex-rs/core/tests/suite/rollout_budget.rs index f955923a2f..99761755e3 100644 --- a/codex-rs/core/tests/suite/rollout_budget.rs +++ b/codex-rs/core/tests/suite/rollout_budget.rs @@ -1,12 +1,14 @@ use anyhow::Result; use codex_core::TurnInputRequest; use codex_core::config::RolloutBudgetConfig; +use codex_extension_api::ExtensionRegistryBuilder; use codex_features::Feature; use codex_model_provider_info::built_in_model_providers; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; use codex_protocol::user_input::UserInput; +use core_test_support::ThreadIdle; use core_test_support::responses::ResponsesRequest; use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; @@ -22,6 +24,7 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use serde_json::json; +use std::sync::Arc; use std::time::Duration; use test_case::test_case; use tokio::time::timeout; @@ -482,7 +485,10 @@ async fn restates_the_current_remainder_after_rollback() -> Result<()> { ], ) .await; + let mut extensions = ExtensionRegistryBuilder::new(); + extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle)); let test = test_codex() + .with_extensions(Arc::new(extensions.build())) .with_config(|config| { config.rollout_budget = Some(RolloutBudgetConfig { reminder_at_remaining_tokens: vec![50], @@ -493,10 +499,14 @@ async fn restates_the_current_remainder_after_rollback() -> Result<()> { .await?; test.submit_turn("rolled-back turn").await?; + ThreadIdle::wait(&test.codex).await; test.codex .submit(Op::ThreadRollback { num_turns: 1 }) .await?; wait_for_event(&test.codex, |event| { + if let EventMsg::Error(error) = event { + panic!("rollback failed: {error:?}"); + } matches!(event, EventMsg::ThreadRolledBack(_)) }) .await;