diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index e57c1ed375..9320030354 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -455,6 +455,18 @@ async fn ingest_turn_prerequisites( ) { if include_initialize { ingest_initialize(reducer, out).await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-2", /*ephemeral*/ false, "gpt-5", + )), + }, + out, + ) + .await; + out.clear(); } reducer @@ -1442,6 +1454,10 @@ fn turn_event_serializes_expected_shape() { runtime: sample_runtime_metadata(), submission_type: None, ephemeral: false, + thread_source: Some("user".to_string()), + initialization_mode: ThreadInitializationMode::New, + subagent_source: None, + parent_thread_id: None, model: Some("gpt-5".to_string()), model_provider: "openai".to_string(), sandbox_policy: Some("read_only"), @@ -1499,6 +1515,10 @@ fn turn_event_serializes_expected_shape() { "runtime_arch": "aarch64" }, "ephemeral": false, + "thread_source": "user", + "initialization_mode": "new", + "subagent_source": null, + "parent_thread_id": null, "model": "gpt-5", "model_provider": "openai", "sandbox_policy": "read_only", @@ -1598,6 +1618,10 @@ async fn accepted_turn_steer_emits_expected_event() { payload["event_params"]["runtime"]["codex_rs_version"], json!("0.1.0") ); + assert_eq!(payload["event_params"]["thread_source"], json!("user")); + assert_eq!(payload["event_params"]["initialization_mode"], json!("new")); + assert_eq!(payload["event_params"]["subagent_source"], json!(null)); + assert_eq!(payload["event_params"]["parent_thread_id"], json!(null)); assert!(payload["event_params"].get("product_client_id").is_none()); } @@ -1626,6 +1650,10 @@ async fn rejected_turn_steer_uses_request_connection_metadata() { payload["event_params"]["runtime"]["codex_rs_version"], json!("0.1.0") ); + assert_eq!(payload["event_params"]["thread_source"], json!("user")); + assert_eq!(payload["event_params"]["initialization_mode"], json!("new")); + assert_eq!(payload["event_params"]["subagent_source"], json!(null)); + assert_eq!(payload["event_params"]["parent_thread_id"], json!(null)); assert_eq!(payload["event_params"]["result"], json!("rejected")); assert_eq!( payload["event_params"]["rejection_reason"], diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index d804f02f30..ef24e6cee3 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -86,6 +86,13 @@ pub(crate) struct CodexRuntimeMetadata { pub(crate) runtime_arch: String, } +pub(crate) struct ThreadMetadata { + pub(crate) thread_source: Option<&'static str>, + pub(crate) initialization_mode: ThreadInitializationMode, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, +} + #[derive(Serialize)] pub(crate) struct ThreadInitializedEventParams { pub(crate) thread_id: String, @@ -341,6 +348,10 @@ pub(crate) struct CodexTurnEventParams { pub(crate) app_server_client: CodexAppServerClientMetadata, pub(crate) runtime: CodexRuntimeMetadata, pub(crate) ephemeral: bool, + pub(crate) thread_source: Option, + pub(crate) initialization_mode: ThreadInitializationMode, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, pub(crate) model: Option, pub(crate) model_provider: String, pub(crate) sandbox_policy: Option<&'static str>, @@ -390,6 +401,10 @@ pub(crate) struct CodexTurnSteerEventParams { pub(crate) accepted_turn_id: Option, pub(crate) app_server_client: CodexAppServerClientMetadata, pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) thread_source: Option, + pub(crate) initialization_mode: ThreadInitializationMode, + pub(crate) subagent_source: Option, + pub(crate) parent_thread_id: Option, pub(crate) num_input_images: usize, pub(crate) result: TurnSteerResult, pub(crate) rejection_reason: Option, @@ -528,6 +543,7 @@ pub(crate) fn codex_turn_steer_event_params( app_server_client: CodexAppServerClientMetadata, runtime: CodexRuntimeMetadata, tracking: &TrackEventsContext, + thread_metadata: ThreadMetadata, turn_steer: CodexTurnSteerEvent, ) -> CodexTurnSteerEventParams { CodexTurnSteerEventParams { @@ -536,6 +552,10 @@ pub(crate) fn codex_turn_steer_event_params( accepted_turn_id: turn_steer.accepted_turn_id, app_server_client, runtime, + thread_source: thread_metadata.thread_source.map(str::to_string), + initialization_mode: thread_metadata.initialization_mode, + subagent_source: thread_metadata.subagent_source, + parent_thread_id: thread_metadata.parent_thread_id, num_input_images: turn_steer.num_input_images, result: turn_steer.result, rejection_reason: turn_steer.rejection_reason, diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index 352ea1c892..bf61b4ae98 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -16,6 +16,7 @@ use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; +use crate::events::ThreadMetadata; use crate::events::TrackEventRequest; use crate::events::codex_app_metadata; use crate::events::codex_compaction_event_params; @@ -86,12 +87,16 @@ struct ConnectionState { #[derive(Clone)] struct ThreadMetadataState { thread_source: Option<&'static str>, + initialization_mode: ThreadInitializationMode, subagent_source: Option, parent_thread_id: Option, } impl ThreadMetadataState { - fn from_session_source(session_source: &SessionSource) -> Self { + fn from_thread_metadata( + session_source: &SessionSource, + initialization_mode: ThreadInitializationMode, + ) -> Self { let (subagent_source, parent_thread_id) = match session_source { SessionSource::SubAgent(subagent_source) => ( Some(subagent_source_name(subagent_source)), @@ -106,6 +111,7 @@ impl ThreadMetadataState { }; Self { thread_source: thread_source_name(session_source), + initialization_mode, subagent_source, parent_thread_id, } @@ -657,7 +663,8 @@ impl AnalyticsReducer { let Some(connection_state) = self.connections.get(&connection_id) else { return; }; - let thread_metadata = ThreadMetadataState::from_session_source(&thread_source); + let thread_metadata = + ThreadMetadataState::from_thread_metadata(&thread_source, initialization_mode); self.thread_connections .insert(thread_id.clone(), connection_id); self.thread_metadata @@ -759,6 +766,13 @@ impl AnalyticsReducer { let Some(connection_state) = self.connections.get(&connection_id) else { return; }; + let Some(thread_metadata) = self.thread_metadata.get(&pending_request.thread_id) else { + tracing::warn!( + thread_id = %pending_request.thread_id, + "dropping turn steer analytics event: missing thread lifecycle metadata" + ); + return; + }; let tracking = TrackEventsContext { model_slug: String::new(), thread_id: pending_request.thread_id, @@ -781,6 +795,12 @@ impl AnalyticsReducer { connection_state.app_server_client.clone(), connection_state.runtime.clone(), &tracking, + ThreadMetadata { + thread_source: thread_metadata.thread_source, + initialization_mode: thread_metadata.initialization_mode, + subagent_source: thread_metadata.subagent_source.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), + }, turn_steer, ), })); @@ -807,6 +827,24 @@ impl AnalyticsReducer { ) }); let Some((app_server_client, runtime)) = connection_metadata else { + if let Some(connection_id) = turn_state.connection_id { + tracing::warn!( + turn_id, + connection_id, + "dropping turn analytics event: missing connection metadata" + ); + } + return; + }; + let Some(thread_id) = turn_state.thread_id.as_ref() else { + return; + }; + let Some(thread_metadata) = self.thread_metadata.get(thread_id) else { + tracing::warn!( + thread_id, + turn_id, + "dropping turn analytics event: missing thread lifecycle metadata" + ); return; }; out.push(TrackEventRequest::TurnEvent(Box::new( @@ -817,6 +855,7 @@ impl AnalyticsReducer { runtime, turn_id.to_string(), turn_state, + thread_metadata, ), }, ))); @@ -829,6 +868,7 @@ fn codex_turn_event_params( runtime: CodexRuntimeMetadata, turn_id: String, turn_state: &TurnState, + thread_metadata: &ThreadMetadataState, ) -> CodexTurnEventParams { let (Some(thread_id), Some(num_input_images), Some(resolved_config), Some(completed)) = ( turn_state.thread_id.clone(), @@ -867,6 +907,10 @@ fn codex_turn_event_params( runtime, submission_type, ephemeral, + thread_source: thread_metadata.thread_source.map(str::to_string), + initialization_mode: thread_metadata.initialization_mode, + subagent_source: thread_metadata.subagent_source.clone(), + parent_thread_id: thread_metadata.parent_thread_id.clone(), model: Some(model), model_provider, sandbox_policy: Some(sandbox_policy_mode(&sandbox_policy)), diff --git a/codex-rs/app-server/tests/suite/v2/turn_start.rs b/codex-rs/app-server/tests/suite/v2/turn_start.rs index ea3ba02b46..694d0a6eef 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -395,6 +395,17 @@ async fn turn_start_tracks_turn_event_analytics() -> Result<()> { assert_eq!(event["event_params"]["model"], "mock-model"); assert_eq!(event["event_params"]["model_provider"], "mock_provider"); assert_eq!(event["event_params"]["sandbox_policy"], "read_only"); + assert_eq!(event["event_params"]["ephemeral"], false); + assert_eq!(event["event_params"]["thread_source"], "user"); + assert_eq!(event["event_params"]["initialization_mode"], "new"); + assert_eq!( + event["event_params"]["subagent_source"], + serde_json::Value::Null + ); + assert_eq!( + event["event_params"]["parent_thread_id"], + serde_json::Value::Null + ); assert_eq!(event["event_params"]["num_input_images"], 1); assert_eq!(event["event_params"]["status"], "completed"); assert!(event["event_params"]["started_at"].as_u64().is_some());