openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-16 12:13:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Honor thread analytics opt-outs when using shared clients (#44646)

Browse Source 
## 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
This commit is contained in:
priyanshusingh-de
2026-09-10 21:20:44 +00:00
committed by copyberry
parent e25bedc166
commit 60825b4988
8 changed files with 295 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
codex-rs/core/src/codex_delegate.rs
View File

@@ -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(),

80
codex-rs/core/src/codex_delegate_tests.rs
View File

@@ -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::<Vec<_>>();
assert_eq!(events, expected_events);
}
#[tokio::test]
async fn delegate_isolation_does_not_depend_on_attribution() {
let (mut parent_session, parent_ctx, _rx_events) =

5
codex-rs/core/src/codex_thread.rs
View File

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

18
codex-rs/core/src/session/session.rs
View File

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

127
codex-rs/core/src/thread_manager_tests.rs
View File

@@ -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::<AnalyticsEventsClient>()
.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<serde_json::Value> = server
.received_requests()
.await
.expect("analytics requests")
.into_iter()
.filter(|request| request.url.path() == "/codex/analytics-events/events")
.flat_map(|request| {
request.body_json::<serde_json::Value>().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<String> = 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() {

34
codex-rs/core/tests/common/lib.rs
View File

@@ -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<Config> 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<T, F>(codex: &CodexThread, matcher: F) -> T
where
F: Fn(&codex_protocol::protocol::EventMsg) -> Option<T>,

38
codex-rs/core/tests/suite/compact_resume_fork.rs
View File

@@ -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<TempDir>, Config, Arc<ThreadManager>, Arc<CodexThread>) {
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<CodexThread>, 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<CodexThread>) {
@@ -887,6 +902,7 @@ async fn compact_conversation(conversation: &Arc<CodexThread>) {
};
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<CodexThread>) -> std::path::PathBuf {

10
codex-rs/core/tests/suite/rollout_budget.rs
View File

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