mirror of
https://github.com/openai/codex.git
synced 2026-10-17 17:08:55 +00:00
Enrich thread archive analytics with thread context (#39797)
## What changed - Add app-server client, runtime, thread source, and parent thread ID metadata to archive and unarchive analytics events when the reducer has that context. - Omit non-automation feature thread sources and skip unavailable metadata. ## Testing - Cover archive and unarchive event enrichment, filtered feature sources, and threads without cached context. - Verify the app-server archive flow emits the enriched analytics payload. GitOrigin-RevId: 163f9c30721cee0172bb4581b6dcd269e44cd941
This commit is contained in:
@@ -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]
|
||||
|
||||
@@ -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")),
|
||||
|
||||
@@ -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<CodexAppServerClientMetadata>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub(crate) runtime: Option<CodexRuntimeMetadata>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub(crate) thread_source: Option<ThreadSource>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub(crate) parent_thread_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
|
||||
@@ -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) => {
|
||||
|
||||
@@ -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-...-<id>.jsonl).
|
||||
|
||||
Reference in New Issue
Block a user