diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 7e92cf5ae2..dfde8cf3e1 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -140,11 +140,13 @@ use codex_app_server_protocol::SubAgentActivityKind; 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::ThreadItem; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSource as AppServerThreadSource; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus as AppServerThreadStatus; +use codex_app_server_protocol::ThreadUnarchivedNotification; use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnDiffUpdatedNotification; @@ -2127,6 +2129,101 @@ async fn thread_originator_overrides_shared_connection_across_thread_events() { }), ] ); + + reducer + .ingest( + AnalyticsFact::ClientResponse { + connection_id: 7, + request_id: RequestId::Integer(4), + response: Box::new(sample_thread_resume_response_with_source( + "thread-private-source", + /*ephemeral*/ false, + "gpt-5", + AppServerSessionSource::Exec, + Some(AppServerThreadSource::Feature( + "private customer feature label".to_string(), + )), + Some("019ee5cf-4d15-77a2-8023-01a9f79b6e7d".to_string()), + )), + thread_originator: Some(TEST_PRODUCT_CLIENT_ID.to_string()), + }, + &mut events, + ) + .await; + events.clear(); + + for notification in [ + ServerNotification::ThreadArchived(ThreadArchivedNotification { + thread_id: "thread-work".to_string(), + }), + ServerNotification::ThreadUnarchived(ThreadUnarchivedNotification { + thread_id: "thread-default".to_string(), + }), + ServerNotification::ThreadArchived(ThreadArchivedNotification { + thread_id: "thread-private-source".to_string(), + }), + ServerNotification::ThreadUnarchived(ThreadUnarchivedNotification { + thread_id: "thread-without-context".to_string(), + }), + ] { + reducer + .ingest( + AnalyticsFact::Notification(Box::new(notification)), + &mut events, + ) + .await; + } + + let mut archives = serde_json::to_value(&events).expect("serialize archive events"); + for event in archives.as_array_mut().expect("archive events") { + assert!(event["event_params"]["occurred_at_ms"].is_u64()); + event["event_params"] + .as_object_mut() + .expect("archive event params") + .remove("occurred_at_ms"); + } + assert_eq!( + archives, + json!([ + { + "event_type": "codex_thread_archive_event", + "event_params": { + "thread_id": "thread-work", + "action": "archived", + "app_server_client": initialized[0]["event_params"]["app_server_client"], + "runtime": initialized[0]["event_params"]["runtime"], + "thread_source": "user", + }, + }, + { + "event_type": "codex_thread_archive_event", + "event_params": { + "thread_id": "thread-default", + "action": "unarchived", + "app_server_client": initialized[1]["event_params"]["app_server_client"], + "runtime": initialized[1]["event_params"]["runtime"], + "thread_source": "user", + }, + }, + { + "event_type": "codex_thread_archive_event", + "event_params": { + "thread_id": "thread-private-source", + "action": "archived", + "app_server_client": initialized[0]["event_params"]["app_server_client"], + "runtime": initialized[0]["event_params"]["runtime"], + "parent_thread_id": "019ee5cf-4d15-77a2-8023-01a9f79b6e7d", + }, + }, + { + "event_type": "codex_thread_archive_event", + "event_params": { + "thread_id": "thread-without-context", + "action": "unarchived", + }, + } + ]) + ); } #[tokio::test] diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 20f95d9e81..89070bd096 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -448,6 +448,10 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { 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")), diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 88b6870619..a06fb6f05d 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -264,6 +264,14 @@ pub(crate) struct ThreadArchiveEventParams { pub(crate) thread_id: String, pub(crate) action: ThreadArchiveAction, pub(crate) occurred_at_ms: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) app_server_client: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) runtime: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) thread_source: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) parent_thread_id: Option, } #[derive(Serialize)] diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index ee3248484a..59434ad086 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -1735,6 +1735,38 @@ impl AnalyticsReducer { ); } + fn thread_archive_event_params( + &self, + thread_id: String, + action: ThreadArchiveAction, + ) -> ThreadArchiveEventParams { + let thread_state = self.threads.get(&thread_id); + let connection_state = self + .thread_connection_id(&thread_id) + .and_then(|connection_id| self.connections.get(&connection_id)); + let thread_metadata = thread_state.and_then(|thread_state| thread_state.metadata.as_ref()); + + ThreadArchiveEventParams { + thread_id, + action, + occurred_at_ms: now_unix_millis(), + app_server_client: thread_state + .zip(connection_state) + .map(|(thread_state, connection_state)| { + thread_state.app_server_client(connection_state) + }), + runtime: connection_state.map(|connection_state| connection_state.runtime.clone()), + thread_source: thread_metadata + .and_then(|thread_metadata| thread_metadata.thread_source.as_ref()) + .filter(|thread_source| { + !matches!(thread_source, ThreadSource::Feature(feature) if feature != "automation") + }) + .cloned(), + parent_thread_id: thread_metadata + .and_then(|thread_metadata| thread_metadata.parent_thread_id.clone()), + } + } + async fn ingest_notification( &mut self, notification: ServerNotification, @@ -1744,21 +1776,19 @@ impl AnalyticsReducer { ServerNotification::ThreadArchived(notification) => { out.push(TrackEventRequest::ThreadArchive(ThreadArchiveEvent { event_type: "codex_thread_archive_event", - event_params: ThreadArchiveEventParams { - thread_id: notification.thread_id, - action: ThreadArchiveAction::Archived, - occurred_at_ms: now_unix_millis(), - }, + event_params: self.thread_archive_event_params( + notification.thread_id, + ThreadArchiveAction::Archived, + ), })); } ServerNotification::ThreadUnarchived(notification) => { out.push(TrackEventRequest::ThreadArchive(ThreadArchiveEvent { event_type: "codex_thread_archive_event", - event_params: ThreadArchiveEventParams { - thread_id: notification.thread_id, - action: ThreadArchiveAction::Unarchived, - occurred_at_ms: now_unix_millis(), - }, + event_params: self.thread_archive_event_params( + notification.thread_id, + ThreadArchiveAction::Unarchived, + ), })); } ServerNotification::ItemStarted(notification) => { diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 09afcee8d2..48a239cc3e 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -3,6 +3,7 @@ use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::create_fake_rollout; use app_test_support::create_mock_responses_server_repeating_assistant; +use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::RequestId; @@ -12,6 +13,7 @@ use codex_app_server_protocol::ThreadArchivedNotification; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; @@ -112,17 +114,28 @@ async fn thread_archive_rejects_owned_unmaterialized_paginated_descendant() -> R async fn thread_archive_requires_materialized_rollout() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; - MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + MockResponsesConfig::new(&server.uri()) + .with_root_config(&format!(r#"chatgpt_base_url = "{}""#, server.uri())) + .write(codex_home.path())?; + mount_analytics_capture(&server, codex_home.path()).await?; let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) - .build_initialized() + .build() .await?; + mcp.initialize_with_client_info(ClientInfo { + name: "codex_work_desktop".to_string(), + title: None, + version: "0.1.0".to_string(), + }) + .await?; // Start a thread. let ThreadStartResponse { thread, .. } = mcp .start_thread(ThreadStartParams { model: Some("mock-model".to_string()), + thread_source: Some(ThreadSource::User), + service_name: Some("codex_work_desktop".to_string()), ..Default::default() }) .await?; @@ -209,6 +222,40 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { .await??; assert_eq!(archived_notification.thread_id, thread.id); + let event = wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_thread_archive_event" + && event["event_params"]["thread_id"] == thread.id + }) + .await?; + let occurred_at_ms = event["event_params"]["occurred_at_ms"] + .as_u64() + .expect("thread archive analytics must include its producer timestamp"); + assert_eq!( + event, + json!({ + "event_type": "codex_thread_archive_event", + "event_params": { + "thread_id": thread.id, + "action": "archived", + "occurred_at_ms": occurred_at_ms, + "app_server_client": { + "product_client_id": "codex_work_desktop", + "client_name": "codex_work_desktop", + "client_version": "0.1.0", + "rpc_transport": "stdio", + "experimental_api_enabled": true, + }, + "runtime": { + "codex_rs_version": env!("CARGO_PKG_VERSION"), + "runtime_os": std::env::consts::OS, + "runtime_os_version": event["event_params"]["runtime"]["runtime_os_version"], + "runtime_arch": std::env::consts::ARCH, + }, + "thread_source": "user", + }, + }) + ); + // Verify file moved. let archived_directory = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); // The archived file keeps the original filename (rollout-...-.jsonl).