diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 84377c6419..bbf83296ec 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -616,7 +616,9 @@ impl AnalyticsEventsClient { pub fn track_notification(&self, notification: &ServerNotification) { if !matches!( notification, - ServerNotification::ThreadClosed(_) + ServerNotification::ThreadArchived(_) + | ServerNotification::ThreadClosed(_) + | ServerNotification::ThreadUnarchived(_) | ServerNotification::TurnStarted(_) | ServerNotification::TurnCompleted(_) | ServerNotification::TurnDiffUpdated(_) diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 816a981317..327c4ce058 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -34,6 +34,12 @@ 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; use crate::facts::AnalyticsFact; @@ -50,6 +56,7 @@ 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::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartResponse; @@ -375,6 +382,14 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { sample_mcp_tool_call_event("non-plugin-mcp", /*plugin_id*/ None), sample_plugin_used_track_event("non-plugin-used", /*plugin_id*/ None), 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, + }, + }), 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")), @@ -729,7 +744,20 @@ async fn flush_waits_for_preceding_fact_delivery() { #[tokio::test] async fn flush_is_noop_when_analytics_is_disabled() { - AnalyticsEventsClient::disabled().flush().await; + 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] diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 0e6c2fa50e..34a71dc012 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -63,6 +63,7 @@ pub(crate) struct TrackEventsRequest { pub(crate) enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), ThreadInitialized(ThreadInitializedEvent), + ThreadArchive(ThreadArchiveEvent), GuardianReview(Box), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), @@ -187,6 +188,26 @@ pub(crate) struct ThreadInitializedEvent { pub(crate) event_params: ThreadInitializedEventParams, } +#[derive(Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ThreadArchiveAction { + Archived, + Unarchived, +} + +#[derive(Serialize)] +pub(crate) struct ThreadArchiveEventParams { + pub(crate) thread_id: String, + pub(crate) action: ThreadArchiveAction, + pub(crate) occurred_at_ms: u64, +} + +#[derive(Serialize)] +pub(crate) struct ThreadArchiveEvent { + pub(crate) event_type: &'static str, + pub(crate) event_params: ThreadArchiveEventParams, +} + #[derive(Serialize)] pub(crate) struct GuardianReviewEventRequest { pub(crate) event_type: &'static str, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 4aadf724b0..4e5190982b 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -51,6 +51,9 @@ use crate::events::ReviewTrigger; use crate::events::Reviewer; use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; +use crate::events::ThreadArchiveAction; +use crate::events::ThreadArchiveEvent; +use crate::events::ThreadArchiveEventParams; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; use crate::events::ToolItemFailureKind; @@ -98,6 +101,7 @@ use crate::facts::TurnStatus; use crate::facts::TurnSteerRejectionReason; use crate::facts::TurnSteerResult; use crate::facts::TurnTokenUsageFact; +use crate::now_unix_millis; use crate::now_unix_seconds; use crate::option_i64_to_u64; use crate::serialize_enum_as_string; @@ -1572,6 +1576,26 @@ impl AnalyticsReducer { out: &mut Vec, ) { match notification { + 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(), + }, + })); + } + 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(), + }, + })); + } ServerNotification::ItemStarted(notification) => { let Some(item_id) = tracked_tool_item_id(¬ification.item) else { return; diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 2020ef2243..0b4933509b 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/codex-rs/app-server/src/outgoing_message.rs @@ -572,6 +572,13 @@ impl OutgoingMessageSender { } pub(crate) async fn send_server_notification(&self, notification: ServerNotification) { + if matches!( + notification, + ServerNotification::ThreadArchived(_) | ServerNotification::ThreadUnarchived(_) + ) { + self.analytics_events_client + .track_notification(¬ification); + } self.send_server_notification_to_connections(&[], notification) .await; } 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 31630aeff2..09afcee8d2 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -28,10 +28,15 @@ use codex_state::DirectionalThreadSpawnEdgeStatus; use codex_state::StateRuntime; use codex_utils_absolute_path::test_support::PathExt; use pretty_assertions::assert_eq; +use serde_json::Value; +use serde_json::json; use std::path::Path; use tempfile::TempDir; use tokio::time::timeout; +use super::analytics::mount_analytics_capture; +use super::analytics::wait_for_matching_analytics_event; + const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] @@ -336,7 +341,10 @@ async fn thread_archive_archives_spawned_descendants() -> Result<()> { async fn thread_archive_succeeds_when_descendant_archive_fails() -> 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 parent_id = create_fake_rollout( codex_home.path(), @@ -423,7 +431,7 @@ async fn thread_archive_succeeds_when_descendant_archive_fails() -> Result<()> { .await??; archived_ids.push(archived_notification.thread_id); } - assert_eq!(archived_ids, vec![parent_id, grandchild_id]); + assert_eq!(archived_ids, vec![parent_id.clone(), grandchild_id.clone()]); assert!( timeout( @@ -465,6 +473,88 @@ async fn thread_archive_succeeds_when_descendant_archive_fails() -> Result<()> { ); } + let repeated_archive_id = mcp + .send_thread_archive_request(ThreadArchiveParams { + thread_id: parent_id.clone(), + }) + .await?; + let _: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(repeated_archive_id)), + ) + .await??; + + let _: ThreadUnarchiveResponse = mcp + .request(|request_id| ClientRequest::ThreadUnarchive { + request_id, + params: ThreadUnarchiveParams { + thread_id: parent_id.clone(), + }, + }) + .await?; + wait_for_matching_analytics_event(&server, DEFAULT_READ_TIMEOUT, |event| { + event["event_type"] == "codex_thread_archive_event" + && event["event_params"]["thread_id"] == parent_id + && event["event_params"]["action"] == "unarchived" + }) + .await?; + + let requests = server + .received_requests() + .await + .ok_or_else(|| anyhow::anyhow!("wiremock did not record requests"))?; + let mut archive_events = Vec::new(); + for request in requests { + if request.url.path() != "/codex/analytics-events/events" { + continue; + } + let payload: Value = serde_json::from_slice(&request.body)?; + let events = payload["events"] + .as_array() + .ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?; + for event in events + .iter() + .filter(|event| event["event_type"] == "codex_thread_archive_event") + { + for (header, expected) in [ + ("authorization", "Bearer chatgpt-token"), + ("chatgpt-account-id", "account-123"), + ] { + assert_eq!( + request + .headers + .get(header) + .and_then(|value| value.to_str().ok()), + Some(expected) + ); + } + archive_events.push(event.clone()); + } + } + + let expected = [ + (parent_id.as_str(), "archived"), + (grandchild_id.as_str(), "archived"), + (parent_id.as_str(), "unarchived"), + ]; + assert_eq!(archive_events.len(), expected.len()); + for (event, (thread_id, action)) in archive_events.iter().zip(expected) { + 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": action, + "occurred_at_ms": occurred_at_ms, + }, + }) + ); + } + Ok(()) }