Track thread archive analytics (#37199)

## What changed

- Emit `codex_thread_archive_event` when a thread is successfully archived or
  unarchived, including the thread ID, action, and producer timestamp.
- Route archive notifications through the analytics client while preserving
  normal app-server notification delivery.

## Testing

- Extend app-server coverage for archive and unarchive analytics, including
  descendant archives and a repeated archive request that fails without
  producing another event.

GitOrigin-RevId: 0d973c232fd795b7cd6cddd47ecc9446234c44dd
This commit is contained in:
Eddie Chen
2026-08-06 02:30:18 +00:00
committed by copyberry
parent 547080e4d6
commit 70b4653232
6 changed files with 176 additions and 4 deletions
+3 -1
View File
@@ -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(_)
+29 -1
View File
@@ -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]
+21
View File
@@ -63,6 +63,7 @@ pub(crate) struct TrackEventsRequest {
pub(crate) enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
ThreadInitialized(ThreadInitializedEvent),
ThreadArchive(ThreadArchiveEvent),
GuardianReview(Box<GuardianReviewEventRequest>),
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,
+24
View File
@@ -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<TrackEventRequest>,
) {
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(&notification.item) else {
return;
@@ -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(&notification);
}
self.send_server_notification_to_connections(&[], notification)
.await;
}
@@ -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(())
}