mirror of
https://github.com/pchuan98/codex.git
synced 2026-07-01 00:31:56 +08:00
[codex-analytics] thread events (#15690)
- add event for thread initialization - thread/start, thread/fork, thread/resume - feature flagged behind `FeatureFlag::GeneralAnalytics` - does not yet support threads started by subagents PR stack: - --> [[telemetry] thread events #15690](https://github.com/openai/codex/pull/15690) - [[telemetry] subagent events #15915](https://github.com/openai/codex/pull/15915) - [[telemetry] turn events #15591](https://github.com/openai/codex/pull/15591) - [[telemetry] steer events #15697](https://github.com/openai/codex/pull/15697) - [[telemetry] queued prompt data #15804](https://github.com/openai/codex/pull/15804) Sample extracted logs in Codex-backend ``` INFO | 2026-03-29 16:39:37 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:398 | Tracked analytics event codex_thread_initialized thread_id=019d3bf7-9f5f-7f82-9877-6d48d1052531 product_surface=codex product_client_id=CODEX_CLI client_name=codex-tui client_version=0.0.0 rpc_transport=in_process experimental_api_enabled=True codex_rs_version=0.0.0 runtime_os=macos runtime_os_version=26.4.0 runtime_arch=aarch64 model=gpt-5.3-codex ephemeral=False thread_source=user initialization_mode=new subagent_source=None parent_thread_id=None created_at=1774827577 | INFO | 2026-03-29 16:45:46 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:398 | Tracked analytics event codex_thread_initialized thread_id=019d3b84-5731-79d0-9b3b-9c6efe5f5066 product_surface=codex product_client_id=CODEX_CLI client_name=codex-tui client_version=0.0.0 rpc_transport=in_process experimental_api_enabled=True codex_rs_version=0.0.0 runtime_os=macos runtime_os_version=26.4.0 runtime_arch=aarch64 model=gpt-5.3-codex ephemeral=False thread_source=user initialization_mode=resumed subagent_source=None parent_thread_id=None created_at=1774820022 | INFO | 2026-03-29 16:45:49 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:398 | Tracked analytics event codex_thread_initialized thread_id=019d3bfd-4cd6-7c12-a13e-48cef02e8c4d product_surface=codex product_client_id=CODEX_CLI client_name=codex-tui client_version=0.0.0 rpc_transport=in_process experimental_api_enabled=True codex_rs_version=0.0.0 runtime_os=macos runtime_os_version=26.4.0 runtime_arch=aarch64 model=gpt-5.3-codex ephemeral=False thread_source=user initialization_mode=forked subagent_source=None parent_thread_id=None created_at=1774827949 | INFO | 2026-03-29 17:20:29 | codex_backend.routers.analytics_events | analytics_events.track_analytics_events:398 | Tracked analytics event codex_thread_initialized thread_id=019d3c1d-0412-7ed2-ad24-c9c0881a36b0 product_surface=codex product_client_id=CODEX_SERVICE_EXEC client_name=codex_exec client_version=0.0.0 rpc_transport=in_process experimental_api_enabled=True codex_rs_version=0.0.0 runtime_os=macos runtime_os_version=26.4.0 runtime_arch=aarch64 model=gpt-5.3-codex ephemeral=False thread_source=user initialization_mode=new subagent_source=None parent_thread_id=None created_at=1774830027 | ``` Notes - `product_client_id` gets canonicalized in codex-backend - subagent threads are addressed in a following pr
This commit is contained in:
@@ -32,6 +32,7 @@ use codex_app_server_protocol::CancelLoginAccountParams;
|
||||
use codex_app_server_protocol::CancelLoginAccountResponse;
|
||||
use codex_app_server_protocol::CancelLoginAccountStatus;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::ClientResponse;
|
||||
use codex_app_server_protocol::CodexErrorInfo as AppServerCodexErrorInfo;
|
||||
use codex_app_server_protocol::CollaborationModeListParams;
|
||||
use codex_app_server_protocol::CollaborationModeListResponse;
|
||||
@@ -179,6 +180,7 @@ use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_backend_client::Client as BackendClient;
|
||||
use codex_chatgpt::connectors;
|
||||
use codex_cloud_requirements::cloud_requirements_loader;
|
||||
use codex_core::AnalyticsEventsClient;
|
||||
use codex_core::AuthManager;
|
||||
use codex_core::CodexAuth;
|
||||
use codex_core::CodexThread;
|
||||
@@ -403,6 +405,7 @@ pub(crate) struct CodexMessageProcessor {
|
||||
auth_manager: Arc<AuthManager>,
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
arg0_paths: Arg0DispatchPaths,
|
||||
config: Arc<Config>,
|
||||
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
|
||||
@@ -433,6 +436,8 @@ struct ListenerTaskContext {
|
||||
thread_manager: Arc<ThreadManager>,
|
||||
thread_state_manager: ThreadStateManager,
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
general_analytics_enabled: bool,
|
||||
thread_watch_manager: ThreadWatchManager,
|
||||
fallback_model_provider: String,
|
||||
codex_home: PathBuf,
|
||||
@@ -455,6 +460,7 @@ pub(crate) struct CodexMessageProcessorArgs {
|
||||
pub(crate) auth_manager: Arc<AuthManager>,
|
||||
pub(crate) thread_manager: Arc<ThreadManager>,
|
||||
pub(crate) outgoing: Arc<OutgoingMessageSender>,
|
||||
pub(crate) analytics_events_client: AnalyticsEventsClient,
|
||||
pub(crate) arg0_paths: Arg0DispatchPaths,
|
||||
pub(crate) config: Arc<Config>,
|
||||
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
|
||||
@@ -519,6 +525,7 @@ impl CodexMessageProcessor {
|
||||
auth_manager,
|
||||
thread_manager,
|
||||
outgoing,
|
||||
analytics_events_client,
|
||||
arg0_paths,
|
||||
config,
|
||||
cli_overrides,
|
||||
@@ -531,6 +538,7 @@ impl CodexMessageProcessor {
|
||||
auth_manager,
|
||||
thread_manager,
|
||||
outgoing: outgoing.clone(),
|
||||
analytics_events_client,
|
||||
arg0_paths,
|
||||
config,
|
||||
cli_overrides,
|
||||
@@ -2086,6 +2094,8 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -2318,6 +2328,17 @@ impl CodexMessageProcessor {
|
||||
sandbox: config_snapshot.sandbox_policy.into(),
|
||||
reasoning_effort: config_snapshot.reasoning_effort,
|
||||
};
|
||||
if listener_task_context.general_analytics_enabled {
|
||||
listener_task_context
|
||||
.analytics_events_client
|
||||
.track_response(
|
||||
request_id.connection_id.0,
|
||||
ClientResponse::ThreadStart {
|
||||
request_id: request_id.request_id.clone(),
|
||||
response: response.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
listener_task_context
|
||||
.outgoing
|
||||
@@ -3796,6 +3817,15 @@ impl CodexMessageProcessor {
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
if self.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
self.analytics_events_client.track_response(
|
||||
request_id.connection_id.0,
|
||||
ClientResponse::ThreadResume {
|
||||
request_id: request_id.request_id.clone(),
|
||||
response: response.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
}
|
||||
@@ -4403,6 +4433,15 @@ impl CodexMessageProcessor {
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
if self.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
self.analytics_events_client.track_response(
|
||||
request_id.connection_id.0,
|
||||
ClientResponse::ThreadFork {
|
||||
request_id: request_id.request_id.clone(),
|
||||
response: response.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
|
||||
@@ -6984,6 +7023,8 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -7071,6 +7112,8 @@ impl CodexMessageProcessor {
|
||||
thread_manager: Arc::clone(&self.thread_manager),
|
||||
thread_state_manager: self.thread_state_manager.clone(),
|
||||
outgoing: Arc::clone(&self.outgoing),
|
||||
analytics_events_client: self.analytics_events_client.clone(),
|
||||
general_analytics_enabled: self.config.features.enabled(Feature::GeneralAnalytics),
|
||||
thread_watch_manager: self.thread_watch_manager.clone(),
|
||||
fallback_model_provider: self.config.model_provider_id.clone(),
|
||||
codex_home: self.config.codex_home.clone(),
|
||||
@@ -7102,6 +7145,8 @@ impl CodexMessageProcessor {
|
||||
outgoing,
|
||||
thread_manager,
|
||||
thread_state_manager,
|
||||
analytics_events_client: _,
|
||||
general_analytics_enabled: _,
|
||||
thread_watch_manager,
|
||||
fallback_model_provider,
|
||||
codex_home,
|
||||
|
||||
@@ -74,6 +74,7 @@ use codex_app_server_protocol::Result;
|
||||
use codex_app_server_protocol::ServerNotification;
|
||||
use codex_app_server_protocol::ServerRequest;
|
||||
use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_core::AppServerRpcTransport;
|
||||
use codex_core::config::Config;
|
||||
use codex_core::config_loader::CloudRequirementsLoader;
|
||||
use codex_core::config_loader::LoaderOverrides;
|
||||
@@ -393,6 +394,7 @@ fn start_uninitialized(args: InProcessStartArgs) -> InProcessClientHandle {
|
||||
config_warnings: args.config_warnings,
|
||||
session_source: args.session_source,
|
||||
enable_codex_api_key_env: args.enable_codex_api_key_env,
|
||||
rpc_transport: AppServerRpcTransport::InProcess,
|
||||
});
|
||||
let mut thread_created_rx = processor.thread_created_receiver();
|
||||
let mut session = ConnectionSessionState::default();
|
||||
|
||||
@@ -36,6 +36,7 @@ use codex_app_server_protocol::ConfigWarningNotification;
|
||||
use codex_app_server_protocol::JSONRPCMessage;
|
||||
use codex_app_server_protocol::TextPosition as AppTextPosition;
|
||||
use codex_app_server_protocol::TextRange as AppTextRange;
|
||||
use codex_core::AppServerRpcTransport;
|
||||
use codex_core::ExecPolicyError;
|
||||
use codex_core::check_execpolicy_for_warnings;
|
||||
use codex_core::config_loader::ConfigLoadError;
|
||||
@@ -623,6 +624,7 @@ pub async fn run_main_with_transport(
|
||||
config_warnings,
|
||||
session_source,
|
||||
enable_codex_api_key_env: false,
|
||||
rpc_transport: analytics_rpc_transport(transport),
|
||||
});
|
||||
let mut thread_created_rx = processor.thread_created_receiver();
|
||||
let mut running_turn_count_rx = processor.subscribe_running_assistant_turn_count();
|
||||
@@ -846,6 +848,13 @@ pub async fn run_main_with_transport(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn analytics_rpc_transport(transport: AppServerTransport) -> AppServerRpcTransport {
|
||||
match transport {
|
||||
AppServerTransport::Stdio => AppServerRpcTransport::Stdio,
|
||||
AppServerTransport::WebSocket { .. } => AppServerRpcTransport::Websocket,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::LogFormat;
|
||||
|
||||
@@ -56,6 +56,7 @@ use codex_app_server_protocol::experimental_required_message;
|
||||
use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_chatgpt::connectors;
|
||||
use codex_core::AnalyticsEventsClient;
|
||||
use codex_core::AppServerRpcTransport;
|
||||
use codex_core::AuthManager;
|
||||
use codex_core::ThreadManager;
|
||||
use codex_core::config::Config;
|
||||
@@ -159,9 +160,11 @@ pub(crate) struct MessageProcessor {
|
||||
external_agent_config_api: ExternalAgentConfigApi,
|
||||
fs_api: FsApi,
|
||||
auth_manager: Arc<AuthManager>,
|
||||
analytics_events_client: AnalyticsEventsClient,
|
||||
fs_watch_manager: FsWatchManager,
|
||||
config: Arc<Config>,
|
||||
config_warnings: Arc<Vec<ConfigWarningNotification>>,
|
||||
rpc_transport: AppServerRpcTransport,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default)]
|
||||
@@ -186,6 +189,7 @@ pub(crate) struct MessageProcessorArgs {
|
||||
pub(crate) config_warnings: Vec<ConfigWarningNotification>,
|
||||
pub(crate) session_source: SessionSource,
|
||||
pub(crate) enable_codex_api_key_env: bool,
|
||||
pub(crate) rpc_transport: AppServerRpcTransport,
|
||||
}
|
||||
|
||||
impl MessageProcessor {
|
||||
@@ -205,6 +209,7 @@ impl MessageProcessor {
|
||||
config_warnings,
|
||||
session_source,
|
||||
enable_codex_api_key_env,
|
||||
rpc_transport,
|
||||
} = args;
|
||||
let auth_manager = AuthManager::shared_with_external_chatgpt_auth_refresher(
|
||||
config.codex_home.clone(),
|
||||
@@ -242,6 +247,7 @@ impl MessageProcessor {
|
||||
auth_manager: auth_manager.clone(),
|
||||
thread_manager: Arc::clone(&thread_manager),
|
||||
outgoing: outgoing.clone(),
|
||||
analytics_events_client: analytics_events_client.clone(),
|
||||
arg0_paths,
|
||||
config: Arc::clone(&config),
|
||||
cli_overrides: cli_overrides.clone(),
|
||||
@@ -262,7 +268,7 @@ impl MessageProcessor {
|
||||
loader_overrides,
|
||||
cloud_requirements,
|
||||
thread_manager,
|
||||
analytics_events_client,
|
||||
analytics_events_client.clone(),
|
||||
);
|
||||
let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone());
|
||||
let fs_api = FsApi::default();
|
||||
@@ -275,9 +281,11 @@ impl MessageProcessor {
|
||||
external_agent_config_api,
|
||||
fs_api,
|
||||
auth_manager,
|
||||
analytics_events_client,
|
||||
fs_watch_manager,
|
||||
config,
|
||||
config_warnings: Arc::new(config_warnings),
|
||||
rpc_transport,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -547,6 +555,7 @@ impl MessageProcessor {
|
||||
// shared thread when another connected client did not opt into
|
||||
// experimental API). Proposed direction is instance-global first-write-wins
|
||||
// with initialize-time mismatch rejection.
|
||||
let analytics_initialize_params = params.clone();
|
||||
let (experimental_api_enabled, opt_out_notification_methods) =
|
||||
match params.capabilities {
|
||||
Some(capabilities) => (
|
||||
@@ -568,7 +577,7 @@ impl MessageProcessor {
|
||||
session.app_server_client_name = Some(name.clone());
|
||||
session.client_version = Some(version.clone());
|
||||
let originator = name.clone();
|
||||
if let Err(error) = set_default_originator(originator) {
|
||||
if let Err(error) = set_default_originator(originator.clone()) {
|
||||
match error {
|
||||
SetOriginatorError::InvalidHeaderValue => {
|
||||
let error = JSONRPCErrorError {
|
||||
@@ -591,6 +600,14 @@ impl MessageProcessor {
|
||||
}
|
||||
}
|
||||
}
|
||||
if self.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
self.analytics_events_client.track_initialize(
|
||||
connection_id.0,
|
||||
analytics_initialize_params,
|
||||
originator,
|
||||
self.rpc_transport,
|
||||
);
|
||||
}
|
||||
set_default_client_residency_requirement(self.config.enforce_residency.value());
|
||||
let user_agent_suffix = format!("{name}; {version}");
|
||||
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
|
||||
|
||||
@@ -20,6 +20,7 @@ use codex_app_server_protocol::TurnStartParams;
|
||||
use codex_app_server_protocol::TurnStartResponse;
|
||||
use codex_app_server_protocol::UserInput;
|
||||
use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_core::AppServerRpcTransport;
|
||||
use codex_core::config::Config;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_core::config_loader::CloudRequirementsLoader;
|
||||
@@ -246,6 +247,7 @@ fn build_test_processor(
|
||||
config_warnings: Vec::new(),
|
||||
session_source: SessionSource::VSCode,
|
||||
enable_codex_api_key_env: false,
|
||||
rpc_transport: AppServerRpcTransport::Stdio,
|
||||
});
|
||||
(processor, outgoing_rx)
|
||||
}
|
||||
|
||||
@@ -1,10 +1,23 @@
|
||||
use anyhow::Result;
|
||||
use app_test_support::ChatGptAuthFixture;
|
||||
use app_test_support::DEFAULT_CLIENT_NAME;
|
||||
use app_test_support::write_chatgpt_auth;
|
||||
use codex_core::auth::AuthCredentialsStoreMode;
|
||||
use codex_core::config::ConfigBuilder;
|
||||
use codex_core::config::types::OtelExporterKind;
|
||||
use codex_core::config::types::OtelHttpProtocol;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use tokio::time::timeout;
|
||||
use wiremock::Mock;
|
||||
use wiremock::MockServer;
|
||||
use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
const SERVICE_VERSION: &str = "0.0.0-test";
|
||||
|
||||
@@ -65,3 +78,91 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
|
||||
assert_eq!(has_metrics, true);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/codex/analytics-events/events"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.mount(server)
|
||||
.await;
|
||||
|
||||
write_chatgpt_auth(
|
||||
codex_home,
|
||||
ChatGptAuthFixture::new("chatgpt-token")
|
||||
.account_id("account-123")
|
||||
.chatgpt_user_id("user-123")
|
||||
.chatgpt_account_id("account-123"),
|
||||
AuthCredentialsStoreMode::File,
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn wait_for_analytics_payload(
|
||||
server: &MockServer,
|
||||
read_timeout: Duration,
|
||||
) -> Result<Value> {
|
||||
let body = timeout(read_timeout, async {
|
||||
loop {
|
||||
let Some(requests) = server.received_requests().await else {
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
continue;
|
||||
};
|
||||
if let Some(request) = requests.iter().find(|request| {
|
||||
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
|
||||
}) {
|
||||
break request.body.clone();
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(25)).await;
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
|
||||
}
|
||||
|
||||
pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> {
|
||||
let events = payload["events"]
|
||||
.as_array()
|
||||
.ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?;
|
||||
events
|
||||
.iter()
|
||||
.find(|event| event["event_type"] == "codex_thread_initialized")
|
||||
.ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present"))
|
||||
}
|
||||
|
||||
pub(crate) fn assert_basic_thread_initialized_event(
|
||||
event: &Value,
|
||||
thread_id: &str,
|
||||
expected_model: &str,
|
||||
initialization_mode: &str,
|
||||
) {
|
||||
assert_eq!(event["event_params"]["thread_id"], thread_id);
|
||||
assert_eq!(
|
||||
event["event_params"]["app_server_client"]["product_client_id"],
|
||||
DEFAULT_CLIENT_NAME
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["app_server_client"]["client_name"],
|
||||
DEFAULT_CLIENT_NAME
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["app_server_client"]["rpc_transport"],
|
||||
"stdio"
|
||||
);
|
||||
assert_eq!(event["event_params"]["model"], expected_model);
|
||||
assert_eq!(event["event_params"]["ephemeral"], false);
|
||||
assert_eq!(event["event_params"]["thread_source"], "user");
|
||||
assert_eq!(
|
||||
event["event_params"]["subagent_source"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["parent_thread_id"],
|
||||
serde_json::Value::Null
|
||||
);
|
||||
assert_eq!(
|
||||
event["event_params"]["initialization_mode"],
|
||||
initialization_mode
|
||||
);
|
||||
assert!(event["event_params"]["created_at"].as_u64().is_some());
|
||||
}
|
||||
|
||||
@@ -38,6 +38,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
#[tokio::test]
|
||||
@@ -177,6 +182,50 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(
|
||||
codex_home.path(),
|
||||
&server.uri(),
|
||||
&server.uri(),
|
||||
/*general_analytics_enabled*/ true,
|
||||
)?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let conversation_id = create_fake_rollout(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved user message",
|
||||
Some("mock_provider"),
|
||||
/*git_info*/ None,
|
||||
)?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let fork_id = mcp
|
||||
.send_thread_fork_request(ThreadForkParams {
|
||||
thread_id: conversation_id,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let fork_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "forked");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
@@ -249,6 +298,7 @@ async fn thread_fork_surfaces_cloud_requirements_load_errors() -> Result<()> {
|
||||
codex_home.path(),
|
||||
&model_server.uri(),
|
||||
&chatgpt_base_url,
|
||||
/*general_analytics_enabled*/ false,
|
||||
)?;
|
||||
write_chatgpt_auth(
|
||||
codex_home.path(),
|
||||
@@ -509,7 +559,13 @@ fn create_config_toml_with_chatgpt_base_url(
|
||||
codex_home: &Path,
|
||||
server_uri: &str,
|
||||
chatgpt_base_url: &str,
|
||||
general_analytics_enabled: bool,
|
||||
) -> std::io::Result<()> {
|
||||
let general_analytics_toml = if general_analytics_enabled {
|
||||
"\n[features]\ngeneral_analytics = true\n".to_string()
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
let config_toml = codex_home.join("config.toml");
|
||||
std::fs::write(
|
||||
config_toml,
|
||||
@@ -521,6 +577,7 @@ sandbox_mode = "read-only"
|
||||
chatgpt_base_url = "{chatgpt_base_url}"
|
||||
|
||||
model_provider = "mock_provider"
|
||||
{general_analytics_toml}
|
||||
|
||||
[model_providers.mock_provider]
|
||||
name = "Mock provider for test"
|
||||
|
||||
@@ -70,6 +70,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
|
||||
|
||||
@@ -150,6 +155,51 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(
|
||||
codex_home.path(),
|
||||
&server.uri(),
|
||||
&server.uri(),
|
||||
/*general_analytics_enabled*/ true,
|
||||
)?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let conversation_id = create_fake_rollout_with_text_elements(
|
||||
codex_home.path(),
|
||||
"2025-01-05T12-00-00",
|
||||
"2025-01-05T12:00:00Z",
|
||||
"Saved user message",
|
||||
Vec::new(),
|
||||
Some("mock_provider"),
|
||||
/*git_info*/ None,
|
||||
)?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let resume_id = mcp
|
||||
.send_thread_resume_request(ThreadResumeParams {
|
||||
thread_id: conversation_id,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
let resume_resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "gpt-5.2-codex", "resumed");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_resume_returns_rollout_history() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
@@ -1449,6 +1499,7 @@ async fn thread_resume_surfaces_cloud_requirements_load_errors() -> Result<()> {
|
||||
codex_home.path(),
|
||||
&model_server.uri(),
|
||||
&chatgpt_base_url,
|
||||
/*general_analytics_enabled*/ false,
|
||||
)?;
|
||||
write_chatgpt_auth(
|
||||
codex_home.path(),
|
||||
@@ -1827,6 +1878,7 @@ model_provider = "mock_provider"
|
||||
|
||||
[features]
|
||||
personality = true
|
||||
general_analytics = true
|
||||
|
||||
[model_providers.mock_provider]
|
||||
name = "Mock provider for test"
|
||||
@@ -1843,7 +1895,13 @@ fn create_config_toml_with_chatgpt_base_url(
|
||||
codex_home: &std::path::Path,
|
||||
server_uri: &str,
|
||||
chatgpt_base_url: &str,
|
||||
general_analytics_enabled: bool,
|
||||
) -> std::io::Result<()> {
|
||||
let general_analytics_toml = if general_analytics_enabled {
|
||||
"\ngeneral_analytics = true".to_string()
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
let config_toml = codex_home.join("config.toml");
|
||||
std::fs::write(
|
||||
config_toml,
|
||||
@@ -1858,6 +1916,7 @@ model_provider = "mock_provider"
|
||||
|
||||
[features]
|
||||
personality = true
|
||||
{general_analytics_toml}
|
||||
|
||||
[model_providers.mock_provider]
|
||||
name = "Mock provider for test"
|
||||
|
||||
@@ -26,6 +26,7 @@ use pretty_assertions::assert_eq;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use tokio::time::timeout;
|
||||
use wiremock::Mock;
|
||||
@@ -34,6 +35,11 @@ use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path;
|
||||
|
||||
use super::analytics::assert_basic_thread_initialized_event;
|
||||
use super::analytics::enable_analytics_capture;
|
||||
use super::analytics::thread_initialized_event;
|
||||
use super::analytics::wait_for_analytics_payload;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
#[tokio::test]
|
||||
@@ -153,6 +159,73 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(
|
||||
codex_home.path(),
|
||||
&server.uri(),
|
||||
&server.uri(),
|
||||
/*general_analytics_enabled*/ true,
|
||||
)?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let req_id = mcp
|
||||
.send_thread_start_request(ThreadStartParams::default())
|
||||
.await?;
|
||||
let resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
|
||||
)
|
||||
.await??;
|
||||
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
|
||||
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
|
||||
let event = thread_initialized_event(&payload)?;
|
||||
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_does_not_track_thread_initialized_analytics_without_feature() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
|
||||
let codex_home = TempDir::new()?;
|
||||
create_config_toml_with_chatgpt_base_url(
|
||||
codex_home.path(),
|
||||
&server.uri(),
|
||||
&server.uri(),
|
||||
/*general_analytics_enabled*/ false,
|
||||
)?;
|
||||
enable_analytics_capture(&server, codex_home.path()).await?;
|
||||
|
||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let req_id = mcp
|
||||
.send_thread_start_request(ThreadStartParams::default())
|
||||
.await?;
|
||||
let resp: JSONRPCResponse = timeout(
|
||||
DEFAULT_READ_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
|
||||
)
|
||||
.await??;
|
||||
let _ = to_response::<ThreadStartResponse>(resp)?;
|
||||
|
||||
let payload = wait_for_analytics_payload(&server, Duration::from_millis(250)).await;
|
||||
assert!(
|
||||
payload.is_err(),
|
||||
"thread analytics should be gated off when general_analytics is disabled"
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
|
||||
let server = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
@@ -455,6 +528,7 @@ async fn thread_start_surfaces_cloud_requirements_load_errors() -> Result<()> {
|
||||
codex_home.path(),
|
||||
&model_server.uri(),
|
||||
&chatgpt_base_url,
|
||||
/*general_analytics_enabled*/ false,
|
||||
)?;
|
||||
write_chatgpt_auth(
|
||||
codex_home.path(),
|
||||
@@ -538,7 +612,13 @@ fn create_config_toml_with_chatgpt_base_url(
|
||||
codex_home: &Path,
|
||||
server_uri: &str,
|
||||
chatgpt_base_url: &str,
|
||||
general_analytics_enabled: bool,
|
||||
) -> std::io::Result<()> {
|
||||
let general_analytics_toml = if general_analytics_enabled {
|
||||
"\n[features]\ngeneral_analytics = true\n".to_string()
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
let config_toml = codex_home.join("config.toml");
|
||||
std::fs::write(
|
||||
config_toml,
|
||||
@@ -550,6 +630,7 @@ sandbox_mode = "read-only"
|
||||
chatgpt_base_url = "{chatgpt_base_url}"
|
||||
|
||||
model_provider = "mock_provider"
|
||||
{general_analytics_toml}
|
||||
|
||||
[model_providers.mock_provider]
|
||||
name = "Mock provider for test"
|
||||
|
||||
Reference in New Issue
Block a user