use super::AnalyticsEventsClient; use super::AnalyticsEventsDestination; use super::AnalyticsEventsQueue; use super::AnalyticsEventsQueueMessage; #[cfg(debug_assertions)] use super::capture_track_events_request; use super::send_track_events; #[cfg(debug_assertions)] use super::send_track_events_request; use super::track_event_request_batches; #[cfg(debug_assertions)] use crate::events::AppServerRpcTransport; use crate::events::CodexAcceptedLineFingerprintsEventParams; use crate::events::CodexAcceptedLineFingerprintsEventRequest; #[cfg(debug_assertions)] use crate::events::CodexAppServerClientMetadata; #[cfg(debug_assertions)] use crate::events::CodexMcpToolCallEventParams; #[cfg(debug_assertions)] use crate::events::CodexMcpToolCallEventRequest; #[cfg(debug_assertions)] use crate::events::CodexPluginMeasurementEventParams; #[cfg(debug_assertions)] use crate::events::CodexPluginMeasurementEventRequest; #[cfg(debug_assertions)] use crate::events::CodexPluginMetadata; #[cfg(debug_assertions)] use crate::events::CodexPluginUsedEventRequest; #[cfg(debug_assertions)] use crate::events::CodexPluginUsedMetadata; #[cfg(debug_assertions)] use crate::events::CodexRuntimeMetadata; #[cfg(debug_assertions)] use crate::events::CodexToolItemEventBase; #[cfg(debug_assertions)] use crate::events::FinalApprovalOutcome; use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; #[cfg(debug_assertions)] use crate::events::ThreadArchiveAction; #[cfg(debug_assertions)] use crate::events::ThreadArchiveEvent; #[cfg(debug_assertions)] use crate::events::ThreadArchiveEventParams; #[cfg(debug_assertions)] use crate::events::ToolItemTerminalStatus; use crate::events::TrackEventRequest; #[cfg(debug_assertions)] use crate::events::codex_artifact_operation_event_request; use crate::facts::AnalyticsFact; use crate::facts::AppInvocation; #[cfg(debug_assertions)] use crate::facts::ArtifactOperation; #[cfg(debug_assertions)] use crate::facts::ArtifactOperationLifecycle; use crate::facts::CustomAnalyticsFact; use crate::facts::ElicitationType; use crate::facts::InvocationType; use crate::facts::PluginMeasurementRow; use crate::facts::PluginMeasurementsInput; use crate::facts::TrackEventsContext; #[cfg(debug_assertions)] use crate::product_attribution::ThreadProducts; use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::CommandExecutionOutputDeltaNotification; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SessionSource as AppServerSessionSource; use codex_app_server_protocol::Thread; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadArchivedNotification; use codex_app_server_protocol::ThreadForkResponse; use codex_app_server_protocol::ThreadRealtimeClosedNotification; use codex_app_server_protocol::ThreadRealtimeItemAddedNotification; use codex_app_server_protocol::ThreadRealtimeStartedNotification; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnDiffUpdatedNotification; use codex_app_server_protocol::TurnInterruptParams; use codex_app_server_protocol::TurnInterruptResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus as AppServerTurnStatus; use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; use codex_http_client::HttpClientFactory; use codex_http_client::OutboundProxyPolicy; #[cfg(debug_assertions)] use codex_login::AuthManager; use codex_model_provider::ChatgptBaseUrl; use codex_protocol::protocol::RealtimeConversationVersion; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; use std::collections::BTreeMap; use std::collections::HashSet; #[cfg(debug_assertions)] use std::fs; #[cfg(debug_assertions)] use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; #[cfg(debug_assertions)] use std::time::SystemTime; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; #[path = "client_product_tests.rs"] mod product_tests; #[cfg(debug_assertions)] impl AnalyticsEventsClient { pub(crate) fn new_for_capture_file(auth_manager: Arc, path: PathBuf) -> Self { Self { queue: Some(AnalyticsEventsQueue::new( auth_manager, AnalyticsEventsDestination::CaptureFile { path }, )), } } } fn sample_accepted_line_fingerprint_event(thread_id: &str) -> TrackEventRequest { TrackEventRequest::AcceptedLineFingerprints(Box::new( CodexAcceptedLineFingerprintsEventRequest { event_type: "codex_accepted_line_fingerprints", event_params: CodexAcceptedLineFingerprintsEventParams { event_type: "codex.accepted_line_fingerprints", turn_id: "turn-1".to_string(), thread_id: thread_id.to_string(), product_surface: Some("codex".to_string()), model_slug: Some("gpt-5.1-codex".to_string()), completed_at: 1, repo_hash: None, accepted_added_lines: 1, accepted_deleted_lines: 0, line_fingerprints: [], }, }, )) } fn sample_skill_track_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest { TrackEventRequest::SkillInvocation(SkillInvocationEventRequest { event_type: "skill_invocation", skill_id: format!("skill-{thread_id}"), skill_name: "doc".to_string(), event_params: SkillInvocationEventParams { product_client_id: None, skill_scope: None, plugin_id: plugin_id.map(str::to_string), remote_plugin_id: None, thread_id: Some(thread_id.to_string()), turn_id: Some("turn-1".to_string()), voice_session_id: None, invoke_type: Some(InvocationType::Explicit), model_slug: Some("gpt-5.1-codex".to_string()), }, }) } #[cfg(debug_assertions)] fn sample_artifact_operation_event(thread_id: &str) -> TrackEventRequest { TrackEventRequest::ArtifactOperation(codex_artifact_operation_event_request( TrackEventsContext { turn_metadata: None, model_slug: "gpt-5.1-codex".to_string(), thread_id: thread_id.to_string(), turn_id: "turn-1".to_string(), product_client_id: "codex_desktop".to_string(), }, ArtifactOperation { item_id: format!("item-{thread_id}"), lifecycle: ArtifactOperationLifecycle::Started, occurred_at_ms: 1, plugin_id: "presentations@openai-primary-runtime".to_string(), script_path: "skills/presentations/container_tools/mark_artifact_operation_started.mjs" .to_string(), skill: "presentations".to_string(), artifact_type: "presentation".to_string(), operation_kind: "create".to_string(), expected_output_count: 1, output_format: "pptx".to_string(), execution_backend: "unified_exec".to_string(), }, )) } fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest { sample_skill_track_event(thread_id, /*plugin_id*/ None) } #[cfg(debug_assertions)] fn sample_mcp_tool_call_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest { TrackEventRequest::McpToolCall(CodexMcpToolCallEventRequest { event_type: "codex_mcp_tool_call_event", event_params: CodexMcpToolCallEventParams { base: CodexToolItemEventBase { thread_id: thread_id.to_string(), session_id: format!("session-{thread_id}"), turn_id: "turn-1".to_string(), root_turn_id: None, item_id: format!("item-{thread_id}"), cell_id: None, parent_call_id: None, originating_response_id: None, subsequent_response_id: None, app_server_client: CodexAppServerClientMetadata { product_client_id: "codex_desktop".to_string(), client_name: None, client_version: None, rpc_transport: AppServerRpcTransport::InProcess, experimental_api_enabled: None, }, runtime: CodexRuntimeMetadata { codex_rs_version: "0.0.0".to_string(), runtime_os: "test".to_string(), runtime_os_version: "test".to_string(), runtime_arch: "test".to_string(), }, thread_source: None, subagent_source: None, parent_thread_id: None, tool_name: "search".to_string(), tool_event_type: None, started_at_ms: 1, completed_at_ms: 2, duration_ms: Some(1), execution_duration_ms: Some(1), review_count: 0, guardian_review_count: 0, user_review_count: 0, final_approval_outcome: FinalApprovalOutcome::NotNeeded, terminal_status: ToolItemTerminalStatus::Completed, failure_kind: None, requested_additional_permissions: false, requested_network_access: false, }, mcp_server_name: "sample".to_string(), mcp_tool_name: "search".to_string(), mcp_error_present: false, plugin_id: plugin_id.map(str::to_string), connector_id: None, voice_session_id: None, elicitation_type: None, }, }) } #[cfg(debug_assertions)] fn sample_plugin_used_track_event(thread_id: &str, plugin_id: Option<&str>) -> TrackEventRequest { TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest { event_type: "codex_plugin_used", event_params: CodexPluginUsedMetadata { plugin: CodexPluginMetadata { plugin_id: plugin_id.map(str::to_string), remote_plugin_id: None, plugin_name: Some("sample".to_string()), marketplace_name: Some("test".to_string()), has_skills: Some(true), mcp_server_count: Some(1), connector_ids: Some(vec!["calendar".to_string()]), product_client_id: Some("codex_desktop".to_string()), }, mcp_server_names: Some(vec!["mcp-1".to_string()]), thread_id: Some(thread_id.to_string()), turn_id: Some("turn-1".to_string()), model_slug: Some("gpt-5.1-codex".to_string()), }, }) } #[cfg(debug_assertions)] fn unique_capture_path(name: &str) -> PathBuf { let nonce = SystemTime::now() .duration_since(SystemTime::UNIX_EPOCH) .expect("system clock should be after Unix epoch") .as_nanos(); std::env::temp_dir().join(format!( "codex-analytics-{name}-{}-{nonce}.jsonl", std::process::id() )) } fn client_with_receiver() -> ( AnalyticsEventsClient, mpsc::Receiver, ) { let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { sender, product_generation: Arc::new(std::sync::atomic::AtomicU64::new(0)), app_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), plugin_used_emitted_keys: Arc::new(Mutex::new(HashSet::new())), }; (AnalyticsEventsClient { queue: Some(queue) }, receiver) } #[test] #[cfg(debug_assertions)] fn analytics_destination_uses_explicit_capture_file() { let capture_path = unique_capture_path("destination"); let destination = AnalyticsEventsDestination::from_base_url_and_capture_file( "https://chatgpt.com/backend-api/".into(), Some(capture_path.clone()), ); assert_eq!( destination, AnalyticsEventsDestination::CaptureFile { path: capture_path.clone() } ); assert_eq!( fs::read_to_string(&capture_path).expect("read capture file"), "" ); #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; let mode = fs::metadata(&capture_path) .expect("read capture file metadata") .permissions() .mode(); assert_eq!(mode & 0o777, 0o600); } fs::remove_file(capture_path).expect("remove capture file"); } #[test] fn analytics_destination_uses_http_without_capture_file() { let commercial_url = "https://chatgpt.com/backend-api/codex/analytics-events/events"; let gov_url = "https://gov.chatgpt.com/backend-api/codex/analytics-events/events"; for (base_url, expected_urls) in [ ( ChatgptBaseUrl::Default, [commercial_url, gov_url, commercial_url], ), ( ChatgptBaseUrl::from("https://chatgpt.com/backend-api/"), [commercial_url; 3], ), ( ChatgptBaseUrl::from("https://proxy.example/backend-api/"), ["https://proxy.example/backend-api/codex/analytics-events/events"; 3], ), ] { let destination = AnalyticsEventsDestination::from_base_url_and_capture_file( base_url.clone(), /*capture_file*/ None, ); assert_eq!(destination, AnalyticsEventsDestination::Http { base_url }); for (api_key, expected_url) in ["sk-test", "sk-proj-fedramp-test", "sk-test"] .into_iter() .zip(expected_urls) { let auth = codex_login::CodexAuth::from_api_key(api_key); assert_eq!( destination.url_for_auth(&auth).as_deref(), Some(expected_url) ); } } } #[test] #[cfg(not(debug_assertions))] fn analytics_destination_ignores_capture_file_in_release() { let destination = AnalyticsEventsDestination::from_base_url_and_capture_file( "https://chatgpt.com/backend-api/".into(), Some(std::path::PathBuf::from("ignored.jsonl")), ); assert_eq!( destination, AnalyticsEventsDestination::Http { base_url: "https://chatgpt.com/backend-api/".into() } ); } #[tokio::test] #[cfg(debug_assertions)] async fn capture_file_writes_exact_serialized_request() { let capture_path = unique_capture_path("single"); let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path.clone(), }; let event = sample_regular_track_event("thread-1"); let expected_event = serde_json::to_value(&event).expect("serialize expected event"); let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(); let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); send_track_events_request( &auth, &destination, vec![event], &factory, /*product_sku*/ None, ) .await; let contents = fs::read_to_string(&capture_path).expect("read capture file"); let lines = contents.lines().collect::>(); assert_eq!(lines.len(), 1); let payload: serde_json::Value = serde_json::from_str(lines[0]).expect("parse captured payload"); assert_eq!(payload, serde_json::json!({"events": [expected_event]})); fs::remove_file(capture_path).expect("remove capture file"); } #[tokio::test] #[cfg(debug_assertions)] async fn capture_file_writes_final_batches_as_separate_lines() { let capture_path = unique_capture_path("batches"); let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path.clone(), }; let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(); let events = vec![ sample_regular_track_event("thread-1"), sample_accepted_line_fingerprint_event("thread-2"), sample_regular_track_event("thread-3"), ]; for batch in track_event_request_batches(events) { let factory = HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault); send_track_events_request( &auth, &destination, batch, &factory, /*product_sku*/ None, ) .await; } let contents = fs::read_to_string(&capture_path).expect("read capture file"); let payloads = contents .lines() .map(|line| serde_json::from_str::(line).expect("parse capture line")) .collect::>(); assert_eq!(payloads.len(), 3); assert_eq!(payloads[0]["events"][0]["skill_id"], "skill-thread-1"); assert_eq!( payloads[1]["events"][0]["event_type"], "codex_accepted_line_fingerprints" ); assert_eq!(payloads[2]["events"][0]["skill_id"], "skill-thread-3"); fs::remove_file(capture_path).expect("remove capture file"); } #[tokio::test] #[cfg(debug_assertions)] async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { let capture_path = unique_capture_path("api-key-plugin-events"); let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path.clone(), }; let auth_manager = codex_login::AuthManager::from_auth_for_testing( codex_login::CodexAuth::from_api_key("sk-test"), ); let plugin_measurement = |thread_id: &str, plugin_id: &str| { TrackEventRequest::PluginMeasurement(CodexPluginMeasurementEventRequest { event_type: "codex_plugin_measurement_event", event_params: CodexPluginMeasurementEventParams { model_slug: None, reasoning_effort: None, thread_id: thread_id.to_string(), turn_id: "turn-1".to_string(), item_id: "item-1".to_string(), originator: "codex_cli_rs".to_string(), plugin_id: plugin_id.to_string(), execution_id: "execution-1".to_string(), operation: "security_scan".to_string(), measurement_name: "findings".to_string(), number_value: 1.0, dimensions: None, }, }) }; send_track_events( &auth_manager, &destination, vec![ sample_regular_track_event("non-plugin-skill"), sample_mcp_tool_call_event("non-plugin-mcp", /*plugin_id*/ None), sample_plugin_used_track_event("non-plugin-used", /*plugin_id*/ None), plugin_measurement("non-plugin-measurement", /*plugin_id*/ ""), sample_accepted_line_fingerprint_event("other-event"), TrackEventRequest::ThreadArchive(ThreadArchiveEvent { event_type: "codex_thread_archive_event", event_params: ThreadArchiveEventParams { thread_id: "non-plugin-thread-archive".to_string(), action: ThreadArchiveAction::Archived, occurred_at_ms: 1, app_server_client: None, runtime: None, thread_source: None, parent_thread_id: None, }, }), sample_plugin_used_track_event("plugin-used", Some("sample@test")), sample_skill_track_event("plugin-skill", Some("sample@test")), sample_mcp_tool_call_event("plugin-mcp", Some("sample@test")), sample_artifact_operation_event("plugin-artifact"), plugin_measurement("plugin-measurement", "sample@test"), ], &ThreadProducts::default(), ) .await; let contents = fs::read_to_string(&capture_path).expect("read capture file"); let lines = contents.lines().collect::>(); assert_eq!(lines.len(), 1); let payload: serde_json::Value = serde_json::from_str(lines[0]).expect("parse captured payload"); let events = payload["events"].as_array().expect("events array"); for event in events { let event_params = event["event_params"].as_object().expect("event params"); for server_owned_field in [ "auth_mode", "api_organization_id", "api_project_id", "api_key_tracking_id", ] { assert!(!event_params.contains_key(server_owned_field)); } } let delivered_events = events .iter() .map(|event| { serde_json::json!({ "event_type": event["event_type"], "plugin_id": event["event_params"]["plugin_id"], "thread_id": event["event_params"]["thread_id"], }) }) .collect::>(); assert_eq!( delivered_events, vec![ serde_json::json!({ "event_type": "codex_plugin_used", "plugin_id": "sample@test", "thread_id": "plugin-used", }), serde_json::json!({ "event_type": "skill_invocation", "plugin_id": "sample@test", "thread_id": "plugin-skill", }), serde_json::json!({ "event_type": "codex_mcp_tool_call_event", "plugin_id": "sample@test", "thread_id": "plugin-mcp", }), serde_json::json!({ "event_type": "codex_artifact_operation", "plugin_id": "presentations@openai-primary-runtime", "thread_id": "plugin-artifact", }), serde_json::json!({ "event_type": "codex_plugin_measurement_event", "plugin_id": "sample@test", "thread_id": "plugin-measurement", }), ] ); fs::remove_file(capture_path).expect("remove capture file"); } #[test] #[cfg(debug_assertions)] fn capture_write_failure_still_consumes_delivery() { let capture_path = unique_capture_path("missing-parent").join("events.jsonl"); let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path }; let payload = crate::events::TrackEventsRequest { events: vec![sample_regular_track_event("thread-1")], }; assert!(capture_track_events_request(&destination, &payload)); } fn sample_turn_start_request() -> ClientRequest { ClientRequest::TurnStart { request_id: RequestId::Integer(1), params: TurnStartParams { thread_id: "thread-1".to_string(), client_user_message_id: None, input: Vec::new(), ..Default::default() }, } } fn sample_turn_steer_request() -> ClientRequest { ClientRequest::TurnSteer { request_id: RequestId::Integer(2), params: TurnSteerParams { thread_id: "thread-1".to_string(), expected_turn_id: "turn-1".to_string(), client_user_message_id: None, input: Vec::new(), responsesapi_client_metadata: None, responsesapi_client_metadata_patch: None, additional_context: None, }, } } fn sample_turn_interrupt_request(turn_id: &str) -> ClientRequest { ClientRequest::TurnInterrupt { request_id: RequestId::Integer(3), params: TurnInterruptParams { thread_id: "thread-1".to_string(), turn_id: turn_id.to_string(), }, } } fn sample_turn_interrupt_response() -> ClientResponsePayload { ClientResponsePayload::TurnInterrupt(TurnInterruptResponse {}) } fn sample_thread_archive_request() -> ClientRequest { ClientRequest::ThreadArchive { request_id: RequestId::Integer(3), params: ThreadArchiveParams { thread_id: "thread-1".to_string(), }, } } fn sample_thread(thread_id: &str) -> Thread { Thread { originator: None, environments: None, id: thread_id.to_string(), extra: None, session_id: format!("session-{thread_id}"), forked_from_id: None, parent_thread_id: None, preview: "first prompt".to_string(), ephemeral: false, section: None, section_entered_at: None, project_id: None, daybreak_enabled: None, history_mode: Default::default(), model_provider: "openai".to_string(), model: None, reasoning_effort: None, created_at: 1, updated_at: 2, recency_at: Some(2), status: AppServerThreadStatus::Idle, path: None, cwd: test_path_buf("/tmp").abs(), cli_version: "0.0.0".to_string(), source: AppServerSessionSource::Exec, can_accept_direct_input: None, thread_source: None, agent_nickname: None, agent_role: None, git_info: None, name: None, turns: Vec::new(), } } fn sample_thread_start_response() -> ClientResponsePayload { ClientResponsePayload::ThreadStart(ThreadStartResponse { disabled_plugin_ids: Vec::new(), thread: sample_thread("thread-1"), model: "gpt-5".to_string(), model_provider: "openai".to_string(), service_tier: None, cwd: test_path_buf("/tmp").abs(), runtime_workspace_roots: Vec::new(), instruction_sources: Vec::new(), approval_policy: AppServerAskForApproval::OnRequest, approvals_reviewer: AppServerApprovalsReviewer::User, sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, multi_agent_mode: Default::default(), }) } fn sample_thread_resume_response() -> ClientResponsePayload { ClientResponsePayload::ThreadResume(ThreadResumeResponse { disabled_plugin_ids: Vec::new(), thread: sample_thread("thread-2"), model: "gpt-5".to_string(), model_provider: "openai".to_string(), service_tier: None, cwd: test_path_buf("/tmp").abs(), runtime_workspace_roots: Vec::new(), instruction_sources: Vec::new(), approval_policy: AppServerAskForApproval::OnRequest, approvals_reviewer: AppServerApprovalsReviewer::User, sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, collaboration_mode: None, multi_agent_mode: Default::default(), initial_turns_page: None, turns_backwards_cursor: None, items_backwards_cursor: None, }) } fn sample_thread_fork_response() -> ClientResponsePayload { ClientResponsePayload::ThreadFork(ThreadForkResponse { disabled_plugin_ids: Vec::new(), thread: sample_thread("thread-3"), model: "gpt-5".to_string(), model_provider: "openai".to_string(), service_tier: None, cwd: test_path_buf("/tmp").abs(), runtime_workspace_roots: Vec::new(), instruction_sources: Vec::new(), approval_policy: AppServerAskForApproval::OnRequest, approvals_reviewer: AppServerApprovalsReviewer::User, sandbox: AppServerSandboxPolicy::DangerFullAccess, active_permission_profile: None, reasoning_effort: None, multi_agent_mode: Default::default(), }) } fn sample_turn_start_response() -> ClientResponsePayload { ClientResponsePayload::TurnStart(TurnStartResponse { turn: Turn { id: "turn-1".to_string(), root_turn_id: None, items_view: codex_app_server_protocol::TurnItemsView::Full, items: Vec::new(), status: AppServerTurnStatus::InProgress, error: None, started_at: None, completed_at: None, duration_ms: None, }, }) } fn sample_turn_steer_response() -> ClientResponsePayload { ClientResponsePayload::TurnSteer(TurnSteerResponse { turn_id: "turn-2".to_string(), }) } #[test] fn track_plugin_measurements_rejects_unbounded_inputs_before_queueing() { let (client, mut receiver) = client_with_receiver(); let measurements = |row_count| PluginMeasurementsInput { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), item_id: "item-1".to_string(), originator: "codex_cli_rs".to_string(), model_slug: None, reasoning_effort: None, plugin_id: "sample@openai-curated".to_string(), execution_id: "execution-1".to_string(), operation: "security_scan".to_string(), rows: vec![ PluginMeasurementRow { measurement_name: "finding_count".to_string(), number_value: 1.0, dimensions: BTreeMap::new(), }; row_count ], }; client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH + 1)); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); let mut oversized_operation = measurements(1); oversized_operation.operation = "o".repeat(65); client.track_plugin_measurements(oversized_operation); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); let mut mixed_rows = measurements(4); mixed_rows.rows[0].measurement_name = "m".repeat(65); mixed_rows.rows[1] .dimensions .insert("d".repeat(65), "valid".to_string()); mixed_rows.rows[2] .dimensions .insert("valid".to_string(), "v".repeat(65)); client.track_plugin_measurements(mixed_rows); assert!(matches!( receiver.try_recv(), Ok(AnalyticsEventsQueueMessage::Fact(fact)) if matches!( fact.as_ref(), AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input)) if input.rows.len() == 1 && input.rows[0].measurement_name == "finding_count" ) )); client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH)); assert!(matches!( receiver.try_recv(), Ok(AnalyticsEventsQueueMessage::Fact(fact)) if matches!( fact.as_ref(), AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input)) if input.rows.len() == MAX_PLUGIN_MEASUREMENTS_PER_BATCH ) )); } #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); for (request_id, request) in [ (RequestId::Integer(1), sample_turn_start_request()), (RequestId::Integer(2), sample_turn_steer_request()), ] { client.track_request(/*connection_id*/ 7, request_id, &request); assert!(matches!( receiver.try_recv(), Ok(AnalyticsEventsQueueMessage::Fact(input)) if matches!(*input, AnalyticsFact::ClientRequest { .. }) )); } client.track_request( /*connection_id*/ 7, RequestId::Integer(3), &sample_turn_interrupt_request("turn-1"), ); assert!(matches!( receiver.try_recv(), Ok(AnalyticsEventsQueueMessage::Fact(input)) if matches!( *input, AnalyticsFact::ExplicitClientInterruptRequest { ref turn_id, requested_at_ms, .. } if turn_id == "turn-1" && requested_at_ms > 0 ) )); let ignored_request = sample_thread_archive_request(); client.track_request( /*connection_id*/ 7, RequestId::Integer(3), &ignored_request, ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); client.track_request( /*connection_id*/ 7, RequestId::Integer(4), &sample_turn_interrupt_request(""), ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[test] fn track_response_only_enqueues_analytics_relevant_responses() { let (client, mut receiver) = client_with_receiver(); for (request_id, response) in [ (RequestId::Integer(1), sample_thread_start_response()), (RequestId::Integer(2), sample_thread_resume_response()), (RequestId::Integer(3), sample_thread_fork_response()), (RequestId::Integer(4), sample_turn_start_response()), (RequestId::Integer(5), sample_turn_steer_response()), (RequestId::Integer(6), sample_turn_interrupt_response()), ] { client.track_response(/*connection_id*/ 7, request_id, &response); assert!(matches!( receiver.try_recv(), Ok(AnalyticsEventsQueueMessage::Fact(input)) if matches!(*input, AnalyticsFact::ClientResponse { .. }) )); } client.track_response( /*connection_id*/ 7, RequestId::Integer(7), &ClientResponsePayload::ThreadArchive(ThreadArchiveResponse {}), ); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[cfg(unix)] #[test] fn track_response_ignores_unserializable_thread_responses() { use std::ffi::OsString; use std::os::unix::ffi::OsStringExt; let (client, mut receiver) = client_with_receiver(); let mut response = sample_thread_start_response(); let ClientResponsePayload::ThreadStart(thread_start) = &mut response else { panic!("expected thread/start response"); }; thread_start.cwd = codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path( std::path::PathBuf::from(OsString::from_vec(vec![b'/', b'b', b'a', b'd', 0xff])), ) .expect("non-UTF-8 Unix paths are valid absolute paths"); client.track_response(/*connection_id*/ 7, RequestId::Integer(1), &response); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[tokio::test] async fn flush_waits_for_preceding_fact_delivery() { let (client, mut receiver) = client_with_receiver(); client.track_request( /*connection_id*/ 7, RequestId::Integer(1), &sample_turn_start_request(), ); let flush = tokio::spawn(async move { client.flush().await }); assert!(matches!( receiver.recv().await, Some(AnalyticsEventsQueueMessage::Fact(input)) if matches!(*input, AnalyticsFact::ClientRequest { .. }) )); let done_tx = match receiver.recv().await { Some(AnalyticsEventsQueueMessage::Flush(done_tx)) => done_tx, _ => panic!("expected analytics flush barrier"), }; tokio::time::sleep(std::time::Duration::from_millis(25)).await; assert!(!flush.is_finished()); done_tx.send(()).expect("flush receiver should remain open"); flush.await.expect("flush task should complete"); } #[tokio::test] async fn flush_is_noop_when_analytics_is_disabled() { let client = AnalyticsEventsClient::new( codex_login::AuthManager::from_auth_for_testing( codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(), ), "https://chatgpt.com/backend-api".to_string(), /*analytics_enabled*/ Some(false), ); client.track_notification(&ServerNotification::ThreadArchived( ThreadArchivedNotification { thread_id: "thread-1".to_string(), }, )); assert!(client.queue.is_none()); client.flush().await; } #[test] fn app_used_preserves_first_classification_and_emits_again_next_turn() { let (client, mut receiver) = client_with_receiver(); let tracking = TrackEventsContext { turn_metadata: None, model_slug: "gpt-5".to_string(), thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), product_client_id: "codex_desktop".to_string(), }; for (turn_id, elicitation_type) in [ ("turn-1", Some(ElicitationType::AuthOrLink)), ("turn-1", None), ("turn-2", None), ] { client.track_app_used( TrackEventsContext { turn_id: turn_id.to_string(), ..tracking.clone() }, AppInvocation { connector_id: Some("calendar".to_string()), app_name: Some("Calendar".to_string()), invocation_type: Some(InvocationType::Implicit), }, elicitation_type, ); } for (turn_id, elicitation_type) in [ ("turn-1", Some(ElicitationType::AuthOrLink)), ("turn-2", None), ] { let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { panic!("expected app-used analytics fact"); }; let AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(input)) = *input else { panic!("expected app-used analytics fact"); }; assert_eq!( ( input.tracking.turn_id.as_str(), input.app.connector_id.as_deref(), input.elicitation_type, ), (turn_id, Some("calendar"), elicitation_type) ); } assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[test] fn track_notification_only_enqueues_analytics_relevant_notifications() { let (client, mut receiver) = client_with_receiver(); let tracked_payload = TurnDiffUpdatedNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), diff: "diff".to_string(), }; let tracked_notification = ServerNotification::TurnDiffUpdated(tracked_payload.clone()); client.track_notification(&tracked_notification); let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { panic!("expected analytics notification"); }; let AnalyticsFact::Notification(notification) = *input else { panic!("expected analytics notification fact"); }; let ServerNotification::TurnDiffUpdated(notification) = *notification else { panic!("expected turn diff notification"); }; assert_eq!(notification, tracked_payload); let ignored_notification = ServerNotification::CommandExecutionOutputDelta(CommandExecutionOutputDeltaNotification { thread_id: "thread-1".to_string(), turn_id: "turn-1".to_string(), item_id: "item-1".to_string(), delta: "output".to_string(), }); client.track_notification(&ignored_notification); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[test] fn track_notification_records_realtime_session_boundaries() { let (client, mut receiver) = client_with_receiver(); client.track_notification(&ServerNotification::ThreadRealtimeStarted( ThreadRealtimeStartedNotification { thread_id: "thread-1".to_string(), realtime_session_id: Some("voice-session-1".to_string()), version: RealtimeConversationVersion::V2, }, )); let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { panic!("expected realtime session started fact"); }; let AnalyticsFact::RealtimeSessionStarted { thread_id, realtime_session_id, started_at, } = *input else { panic!("expected realtime session started fact"); }; assert_eq!(thread_id, "thread-1"); assert_eq!(realtime_session_id.as_deref(), Some("voice-session-1")); assert!(started_at > 0); client.track_notification(&ServerNotification::ThreadRealtimeClosed( ThreadRealtimeClosedNotification { thread_id: "thread-1".to_string(), reason: Some("completed".to_string()), }, )); let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { panic!("expected realtime session closed fact"); }; let AnalyticsFact::RealtimeSessionClosed { thread_id, closed_at, } = *input else { panic!("expected realtime session closed fact"); }; assert_eq!(thread_id, "thread-1"); assert!(closed_at >= started_at); } #[test] fn realtime_handoff_tracks_only_marker_without_transcript() { let (client, mut receiver) = client_with_receiver(); client.track_notification(&ServerNotification::ThreadRealtimeItemAdded( ThreadRealtimeItemAddedNotification { thread_id: "thread-1".to_string(), item: serde_json::json!({ "type": "handoff_request", "input_transcript": "private speech", }), }, )); let Ok(AnalyticsEventsQueueMessage::Fact(input)) = receiver.try_recv() else { panic!("expected realtime handoff marker"); }; assert!(matches!( *input, AnalyticsFact::RealtimeHandoffRequested { thread_id } if thread_id == "thread-1" )); client.track_notification(&ServerNotification::ThreadRealtimeItemAdded( ThreadRealtimeItemAddedNotification { thread_id: "thread-1".to_string(), item: serde_json::json!({"type": "input_transcript"}), }, )); assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } #[test] fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() { let batches = track_event_request_batches(vec![ sample_regular_track_event("thread-1"), sample_regular_track_event("thread-2"), sample_accepted_line_fingerprint_event("thread-3"), sample_accepted_line_fingerprint_event("thread-4"), sample_regular_track_event("thread-5"), sample_regular_track_event("thread-6"), ]); assert_eq!(batches.len(), 4); assert_eq!(batches[0].len(), 2); assert_eq!(batches[1].len(), 1); assert_eq!(batches[2].len(), 1); assert_eq!(batches[3].len(), 2); assert!(batches[1][0].should_send_in_isolated_request()); assert!(batches[2][0].should_send_in_isolated_request()); } #[cfg(debug_assertions)] #[tokio::test] async fn account_switch_between_batches_does_not_send_old_credentials() { use codex_http_client::DestinationPolicy; use codex_http_client::NetworkPolicyController; use codex_login::AuthCredentialsStoreMode as Store; use tokio::io::AsyncReadExt; use tokio::io::AsyncWriteExt; let home = unique_capture_path("batch-account-switch"); fs::create_dir_all(&home).unwrap(); let mut stored: codex_login::AuthDotJson = serde_json::from_value(serde_json::json!({ "auth_mode": "chatgptAuthTokens", "tokens": { "id_token": "e30.e30.sig", "access_token": "test-access-token-a", "refresh_token": "", "account_id": "account-a", }, })) .unwrap(); stored.last_refresh = Some(SystemTime::now().into()); codex_login::save_auth(&home, &stored, Store::File, Default::default()).unwrap(); let controller = NetworkPolicyController::default(); let policy = controller.policy(); let manager = Arc::new( codex_login::AuthManager::new( home.clone(), /*enable_codex_api_key_env*/ false, Store::File, /*forced_chatgpt_workspace_id*/ None, /*chatgpt_base_url*/ None, Default::default(), codex_login::AuthRouteConfig::from_http_client_factory( HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault) .with_network_policy(policy.clone()), ), ) .await, ); assert!(controller.publish(policy.revision(), DestinationPolicy::Unrestricted)); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let destination = AnalyticsEventsDestination::Http { base_url: format!("http://{}", listener.local_addr().unwrap()).into(), }; let requests = Arc::new(Mutex::new(Vec::new())); let captured = Arc::clone(&requests); let server_manager = Arc::clone(&manager); let server_home = home.clone(); let server = tokio::spawn(async move { loop { let (mut stream, _) = listener.accept().await.unwrap(); let mut request = Vec::new(); loop { let mut buffer = [0; 4096]; let count = stream.read(&mut buffer).await.unwrap(); assert!(count > 0); request.extend_from_slice(&buffer[..count]); if let Some(end) = request.windows(4).position(|part| part == b"\r\n\r\n") { let headers = String::from_utf8_lossy(&request[..end]).to_lowercase(); let length: usize = headers .lines() .find_map(|line| line.strip_prefix("content-length: ")) .unwrap() .parse() .unwrap(); if request.len() >= end + 4 + length { break; } } } let first = { let mut requests = captured.lock().unwrap(); requests.push(String::from_utf8(request).unwrap()); requests.len() == 1 }; if first { let tokens = stored.tokens.as_mut().unwrap(); tokens.account_id = Some("account-b".to_string()); tokens.access_token = "test-access-token-b".to_string(); codex_login::save_auth(&server_home, &stored, Store::File, Default::default()) .unwrap(); server_manager.reload().await; assert!(controller.publish(policy.revision(), DestinationPolicy::Unrestricted)); } let _ = stream .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") .await; } }); let events = vec![ sample_accepted_line_fingerprint_event("one"), sample_accepted_line_fingerprint_event("two"), ]; send_track_events(&manager, &destination, events, &ThreadProducts::default()).await; server.abort(); let _ = server.await; let requests = requests.lock().unwrap(); assert_eq!(requests.len(), 1); assert!( requests[0] .to_lowercase() .contains("chatgpt-account-id: account-a") ); assert!(requests[0].contains("test-access-token-a")); fs::remove_dir_all(home).unwrap(); }