diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 0551687fd8..7361cee099 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -7,6 +7,7 @@ use crate::events::CodexPluginEventRequest; use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexRuntimeMetadata; use crate::events::CodexTurnEventRequest; +use crate::events::CodexTurnSteerEventRequest; use crate::events::ThreadInitializationMode; use crate::events::ThreadInitializedEvent; use crate::events::ThreadInitializedEventParams; @@ -14,11 +15,13 @@ use crate::events::TrackEventRequest; use crate::events::codex_app_metadata; use crate::events::codex_plugin_metadata; use crate::events::codex_plugin_used_metadata; +use crate::events::codex_turn_steer_event_params; use crate::events::subagent_thread_started_event_request; use crate::facts::AnalyticsFact; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexTurnSteerEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::InvocationType; use crate::facts::PluginState; @@ -30,6 +33,9 @@ use crate::facts::SubAgentThreadStartedInput; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; +use crate::facts::TurnSteerInput; +use crate::facts::TurnSteerRejectionReason; +use crate::facts::TurnSteerResult; use crate::facts::TurnSubmissionType; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; @@ -243,6 +249,25 @@ fn sample_turn_resolved_config(turn_id: &str) -> TurnResolvedConfigFact { } } +fn sample_app_server_client_metadata() -> CodexAppServerClientMetadata { + CodexAppServerClientMetadata { + product_client_id: "codex-tui".to_string(), + client_name: Some("codex-tui".to_string()), + client_version: Some("1.0.0".to_string()), + rpc_transport: AppServerRpcTransport::Stdio, + experimental_api_enabled: None, + } +} + +fn sample_runtime_metadata() -> CodexRuntimeMetadata { + CodexRuntimeMetadata { + codex_rs_version: "0.1.0".to_string(), + runtime_os: "macos".to_string(), + runtime_os_version: "15.3.1".to_string(), + runtime_arch: "aarch64".to_string(), + } +} + async fn ingest_turn_prerequisites( reducer: &mut AnalyticsReducer, out: &mut Vec, @@ -1065,7 +1090,7 @@ fn turn_event_serializes_expected_shape() { is_first_turn: true, status: Some(TurnStatus::Completed), turn_error: None, - steer_count: None, + steer_count: Some(0), total_tool_call_count: None, shell_command_count: None, file_change_count: None, @@ -1111,7 +1136,7 @@ fn turn_event_serializes_expected_shape() { "is_first_turn": true, "status": "completed", "turn_error": null, - "steer_count": null, + "steer_count": 0, "total_tool_call_count": null, "shell_command_count": null, "file_change_count": null, @@ -1133,6 +1158,280 @@ fn turn_event_serializes_expected_shape() { ); } +#[test] +fn turn_steer_event_serializes_expected_shape() { + let tracking = TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }; + let event = TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest { + event_type: "codex_turn_steer_event", + event_params: codex_turn_steer_event_params( + sample_app_server_client_metadata(), + sample_runtime_metadata(), + &tracking, + CodexTurnSteerEvent { + expected_turn_id: Some("turn-2".to_string()), + accepted_turn_id: Some("turn-2".to_string()), + num_input_images: 2, + result: TurnSteerResult::Accepted, + rejection_reason: None, + created_at: 1_716_000_123, + }, + ), + }); + + let payload = serde_json::to_value(&event).expect("serialize turn steer event"); + + assert_eq!(payload["event_type"], json!("codex_turn_steer_event")); + assert_eq!(payload["event_params"]["thread_id"], json!("thread-2")); + assert_eq!(payload["event_params"]["expected_turn_id"], json!("turn-2")); + assert_eq!(payload["event_params"]["accepted_turn_id"], json!("turn-2")); + assert_eq!( + payload["event_params"]["app_server_client"], + json!({ + "product_client_id": "codex-tui", + "client_name": "codex-tui", + "client_version": "1.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": null, + }) + ); + assert_eq!( + payload["event_params"]["runtime"], + json!({ + "codex_rs_version": "0.1.0", + "runtime_os": "macos", + "runtime_os_version": "15.3.1", + "runtime_arch": "aarch64", + }) + ); + assert!(payload["event_params"].get("product_client_id").is_none()); + assert_eq!(payload["event_params"]["num_input_images"], json!(2)); + assert_eq!(payload["event_params"]["result"], json!("accepted")); + assert_eq!(payload["event_params"]["rejection_reason"], json!(null)); + assert_eq!(payload["event_params"]["started_at"], json!(1_716_000_123)); + assert!(payload["event_params"].get("created_at").is_none()); +} + +#[test] +fn rejected_turn_steer_event_serializes_expected_shape() { + let tracking = TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-3".to_string(), + turn_id: "turn-3".to_string(), + }; + let event = TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest { + event_type: "codex_turn_steer_event", + event_params: codex_turn_steer_event_params( + sample_app_server_client_metadata(), + sample_runtime_metadata(), + &tracking, + CodexTurnSteerEvent { + expected_turn_id: Some("turn-expected".to_string()), + accepted_turn_id: None, + num_input_images: 1, + result: TurnSteerResult::Rejected, + rejection_reason: Some(TurnSteerRejectionReason::ExpectedTurnMismatch), + created_at: 1_716_000_124, + }, + ), + }); + + let payload = serde_json::to_value(&event).expect("serialize rejected turn steer event"); + + assert_eq!(payload["event_type"], json!("codex_turn_steer_event")); + assert_eq!(payload["event_params"]["thread_id"], json!("thread-3")); + assert_eq!( + payload["event_params"]["expected_turn_id"], + json!("turn-expected") + ); + assert_eq!(payload["event_params"]["accepted_turn_id"], json!(null)); + assert_eq!( + payload["event_params"]["app_server_client"]["product_client_id"], + json!("codex-tui") + ); + assert_eq!( + payload["event_params"]["runtime"]["codex_rs_version"], + json!("0.1.0") + ); + assert!(payload["event_params"].get("product_client_id").is_none()); + assert_eq!(payload["event_params"]["num_input_images"], json!(1)); + assert_eq!(payload["event_params"]["result"], json!("rejected")); + assert_eq!( + payload["event_params"]["rejection_reason"], + json!("expected_turn_mismatch") + ); + assert_eq!(payload["event_params"]["started_at"], json!(1_716_000_124)); + assert!(payload["event_params"].get("created_at").is_none()); +} + +#[tokio::test] +async fn turn_steer_uses_connection_metadata() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ false, + /*include_started*/ false, + /*include_token_usage*/ false, + ) + .await; + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: Some("turn-2".to_string()), + accepted_turn_id: Some("turn-2".to_string()), + num_input_images: 2, + result: TurnSteerResult::Accepted, + rejection_reason: None, + created_at: 1_716_000_125, + }, + })), + &mut out, + ) + .await; + + assert_eq!(out.len(), 1); + let payload = serde_json::to_value(&out[0]).expect("serialize turn steer event"); + assert_eq!(payload["event_type"], json!("codex_turn_steer_event")); + assert_eq!( + payload["event_params"]["app_server_client"], + json!({ + "product_client_id": "codex-tui", + "client_name": "codex-tui", + "client_version": "1.0.0", + "rpc_transport": "stdio", + "experimental_api_enabled": null, + }) + ); + assert_eq!( + payload["event_params"]["runtime"], + json!({ + "codex_rs_version": "0.1.0", + "runtime_os": "macos", + "runtime_os_version": "15.3.1", + "runtime_arch": "aarch64", + }) + ); + assert!(payload["event_params"].get("product_client_id").is_none()); +} + +#[tokio::test] +async fn rejected_turn_steer_uses_thread_connection_metadata() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + reducer + .ingest( + AnalyticsFact::Initialize { + connection_id: 7, + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: None, + }, + product_client_id: "codex-tui".to_string(), + runtime: sample_runtime_metadata(), + rpc_transport: AppServerRpcTransport::Stdio, + }, + &mut out, + ) + .await; + reducer + .ingest( + AnalyticsFact::Response { + connection_id: 7, + response: Box::new(sample_thread_start_response( + "thread-2", /*ephemeral*/ false, "gpt-5", + )), + }, + &mut out, + ) + .await; + out.clear(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: String::new(), + thread_id: "thread-2".to_string(), + turn_id: String::new(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: None, + accepted_turn_id: None, + num_input_images: 1, + result: TurnSteerResult::Rejected, + rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn), + created_at: 1_716_000_126, + }, + })), + &mut out, + ) + .await; + + assert_eq!(out.len(), 1); + let payload = serde_json::to_value(&out[0]).expect("serialize turn steer event"); + assert_eq!(payload["event_type"], json!("codex_turn_steer_event")); + assert_eq!( + payload["event_params"]["app_server_client"]["product_client_id"], + json!("codex-tui") + ); + assert_eq!( + payload["event_params"]["runtime"]["codex_rs_version"], + json!("0.1.0") + ); + assert_eq!(payload["event_params"]["result"], json!("rejected")); + assert_eq!( + payload["event_params"]["rejection_reason"], + json!("no_active_turn") + ); +} + +#[tokio::test] +async fn turn_steer_does_not_emit_without_connection_metadata() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: Some("turn-2".to_string()), + accepted_turn_id: None, + num_input_images: 1, + result: TurnSteerResult::Rejected, + rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn), + created_at: 1_716_000_126, + }, + })), + &mut out, + ) + .await; + + assert!(out.is_empty()); +} + #[tokio::test] async fn turn_lifecycle_emits_turn_event() { let mut reducer = AnalyticsReducer::default(); @@ -1170,6 +1469,7 @@ async fn turn_lifecycle_emits_turn_event() { ); assert_eq!(payload["event_params"]["num_input_images"], json!(1)); assert_eq!(payload["event_params"]["status"], json!("completed")); + assert_eq!(payload["event_params"]["steer_count"], json!(0)); assert_eq!(payload["event_params"]["started_at"], json!(455)); assert_eq!(payload["event_params"]["completed_at"], json!(456)); assert_eq!(payload["event_params"]["duration_ms"], json!(1234)); @@ -1183,6 +1483,104 @@ async fn turn_lifecycle_emits_turn_event() { assert_eq!(payload["event_params"]["total_tokens"], json!(321)); } +#[tokio::test] +async fn accepted_steers_increment_turn_steer_count() { + let mut reducer = AnalyticsReducer::default(); + let mut out = Vec::new(); + + ingest_turn_prerequisites( + &mut reducer, + &mut out, + /*include_initialize*/ true, + /*include_resolved_config*/ true, + /*include_started*/ true, + /*include_token_usage*/ false, + ) + .await; + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: Some("turn-2".to_string()), + accepted_turn_id: Some("turn-2".to_string()), + num_input_images: 0, + result: TurnSteerResult::Accepted, + rejection_reason: None, + created_at: 1, + }, + })), + &mut out, + ) + .await; + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: None, + accepted_turn_id: None, + num_input_images: 0, + result: TurnSteerResult::Rejected, + rejection_reason: Some(TurnSteerRejectionReason::NoActiveTurn), + created_at: 2, + }, + })), + &mut out, + ) + .await; + + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer(TurnSteerInput { + tracking: TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-2".to_string(), + turn_id: "turn-2".to_string(), + }, + turn_steer: CodexTurnSteerEvent { + expected_turn_id: Some("turn-2".to_string()), + accepted_turn_id: Some("turn-2".to_string()), + num_input_images: 1, + result: TurnSteerResult::Accepted, + rejection_reason: None, + created_at: 3, + }, + })), + &mut out, + ) + .await; + + reducer + .ingest( + AnalyticsFact::Notification(Box::new(sample_turn_completed_notification( + "thread-2", + "turn-2", + AppServerTurnStatus::Completed, + /*codex_error_info*/ None, + ))), + &mut out, + ) + .await; + + let turn_event = out + .iter() + .find(|event| matches!(event, TrackEventRequest::TurnEvent(_))) + .expect("turn event should be emitted"); + let payload = serde_json::to_value(turn_event).expect("serialize turn event"); + assert_eq!(payload["event_params"]["steer_count"], json!(2)); +} + #[tokio::test] async fn queued_submission_type_emits_queued_turn_event() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 41802bdb0e..5a5d2f30b5 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -6,6 +6,7 @@ use crate::facts::AnalyticsFact; use crate::facts::AppInvocation; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexTurnSteerEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; @@ -14,6 +15,7 @@ use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; +use crate::facts::TurnSteerInput; use crate::facts::TurnTokenUsageFact; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; @@ -203,6 +205,15 @@ impl AnalyticsEventsClient { ))); } + pub fn track_turn_steer(&self, tracking: TrackEventsContext, turn_steer: CodexTurnSteerEvent) { + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnSteer( + TurnSteerInput { + tracking, + turn_steer, + }, + ))); + } + pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { self.record_fact(AnalyticsFact::Custom( CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 48914d9de9..23e6cccbe4 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -1,9 +1,12 @@ use crate::facts::AppInvocation; +use crate::facts::CodexTurnSteerEvent; use crate::facts::InvocationType; use crate::facts::PluginState; use crate::facts::SubAgentThreadStartedInput; use crate::facts::TrackEventsContext; use crate::facts::TurnStatus; +use crate::facts::TurnSteerRejectionReason; +use crate::facts::TurnSteerResult; use crate::facts::TurnSubmissionType; use codex_app_server_protocol::CodexErrorInfo; use codex_login::default_client::originator; @@ -41,6 +44,7 @@ pub(crate) enum TrackEventRequest { AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), TurnEvent(Box), + TurnSteer(CodexTurnSteerEventRequest), PluginUsed(CodexPluginUsedEventRequest), PluginInstalled(CodexPluginEventRequest), PluginUninstalled(CodexPluginEventRequest), @@ -172,6 +176,25 @@ pub(crate) struct CodexTurnEventRequest { pub(crate) event_params: CodexTurnEventParams, } +#[derive(Serialize)] +pub(crate) struct CodexTurnSteerEventParams { + pub(crate) thread_id: String, + pub(crate) expected_turn_id: Option, + pub(crate) accepted_turn_id: Option, + pub(crate) app_server_client: CodexAppServerClientMetadata, + pub(crate) runtime: CodexRuntimeMetadata, + pub(crate) num_input_images: usize, + pub(crate) result: TurnSteerResult, + pub(crate) rejection_reason: Option, + pub(crate) started_at: u64, +} + +#[derive(Serialize)] +pub(crate) struct CodexTurnSteerEventRequest { + pub(crate) event_type: &'static str, + pub(crate) event_params: CodexTurnSteerEventParams, +} + #[derive(Serialize)] pub(crate) struct CodexPluginMetadata { pub(crate) plugin_id: Option, @@ -263,6 +286,25 @@ pub(crate) fn codex_plugin_used_metadata( } } +pub(crate) fn codex_turn_steer_event_params( + app_server_client: CodexAppServerClientMetadata, + runtime: CodexRuntimeMetadata, + tracking: &TrackEventsContext, + turn_steer: CodexTurnSteerEvent, +) -> CodexTurnSteerEventParams { + CodexTurnSteerEventParams { + thread_id: tracking.thread_id.clone(), + expected_turn_id: turn_steer.expected_turn_id, + accepted_turn_id: turn_steer.accepted_turn_id, + app_server_client, + runtime, + num_input_images: turn_steer.num_input_images, + result: turn_steer.result, + rejection_reason: turn_steer.rejection_reason, + started_at: turn_steer.created_at, + } +} + pub(crate) fn thread_source_name(thread_source: &SessionSource) -> Option<&'static str> { match thread_source { SessionSource::Cli | SessionSource::VSCode | SessionSource::Exec => Some("user"), diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index ed5d96bb25..6c11b06c28 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -81,6 +81,33 @@ pub enum TurnStatus { Interrupted, } +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TurnSteerResult { + Accepted, + Rejected, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum TurnSteerRejectionReason { + NoActiveTurn, + ExpectedTurnMismatch, + NonSteerableReview, + NonSteerableCompact, + EmptyInput, +} + +#[derive(Clone)] +pub struct CodexTurnSteerEvent { + pub expected_turn_id: Option, + pub accepted_turn_id: Option, + pub num_input_images: usize, + pub result: TurnSteerResult, + pub rejection_reason: Option, + pub created_at: u64, +} + #[derive(Clone, Debug)] pub struct SkillInvocation { pub skill_name: String, @@ -142,6 +169,7 @@ pub(crate) enum CustomAnalyticsFact { SubAgentThreadStarted(SubAgentThreadStartedInput), TurnResolvedConfig(Box), TurnTokenUsage(Box), + TurnSteer(TurnSteerInput), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), @@ -149,6 +177,11 @@ pub(crate) enum CustomAnalyticsFact { PluginStateChanged(PluginStateChangedInput), } +pub(crate) struct TurnSteerInput { + pub tracking: TrackEventsContext, + pub turn_steer: CodexTurnSteerEvent, +} + pub(crate) struct SkillInvokedInput { pub tracking: TrackEventsContext, pub invocations: Vec, diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 5a4630ff47..b3e829ea03 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -6,12 +6,15 @@ mod reducer; pub use client::AnalyticsEventsClient; pub use events::AppServerRpcTransport; pub use facts::AppInvocation; +pub use facts::CodexTurnSteerEvent; pub use facts::InvocationType; pub use facts::SkillInvocation; pub use facts::SubAgentThreadStartedInput; pub use facts::TrackEventsContext; pub use facts::TurnResolvedConfigFact; pub use facts::TurnStatus; +pub use facts::TurnSteerRejectionReason; +pub use facts::TurnSteerResult; pub use facts::TurnSubmissionType; pub use facts::TurnTokenUsageFact; pub use facts::build_track_events_context; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index e24900ef04..d6f3ebd5e5 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -7,6 +7,7 @@ use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexRuntimeMetadata; use crate::events::CodexTurnEventParams; use crate::events::CodexTurnEventRequest; +use crate::events::CodexTurnSteerEventRequest; use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; use crate::events::ThreadInitializationMode; @@ -16,20 +17,25 @@ use crate::events::TrackEventRequest; use crate::events::codex_app_metadata; use crate::events::codex_plugin_metadata; use crate::events::codex_plugin_used_metadata; +use crate::events::codex_turn_steer_event_params; use crate::events::plugin_state_event_type; use crate::events::subagent_thread_started_event_request; use crate::events::thread_source_name; use crate::facts::AnalyticsFact; use crate::facts::AppMentionedInput; use crate::facts::AppUsedInput; +use crate::facts::CodexTurnSteerEvent; use crate::facts::CustomAnalyticsFact; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; use crate::facts::PluginUsedInput; use crate::facts::SkillInvokedInput; use crate::facts::SubAgentThreadStartedInput; +use crate::facts::TrackEventsContext; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnStatus; +use crate::facts::TurnSteerInput; +use crate::facts::TurnSteerResult; use crate::facts::TurnTokenUsageFact; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponse; @@ -57,6 +63,7 @@ pub(crate) struct AnalyticsReducer { requests: HashMap<(u64, RequestId), RequestState>, turns: HashMap, connections: HashMap, + thread_connections: HashMap, } struct ConnectionState { @@ -89,6 +96,7 @@ struct TurnState { started_at: Option, token_usage: Option, completed: Option, + steer_count: usize, } impl AnalyticsReducer { @@ -135,6 +143,9 @@ impl AnalyticsReducer { CustomAnalyticsFact::TurnTokenUsage(input) => { self.ingest_turn_token_usage(*input, out); } + CustomAnalyticsFact::TurnSteer(input) => { + self.ingest_turn_steer(input, out); + } CustomAnalyticsFact::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; } @@ -198,6 +209,8 @@ impl AnalyticsReducer { let ClientRequest::TurnStart { params, .. } = request else { return; }; + self.thread_connections + .insert(params.thread_id.clone(), connection_id); self.requests.insert( (connection_id, request_id), RequestState::TurnStart(PendingTurnStartState { @@ -229,6 +242,7 @@ impl AnalyticsReducer { started_at: None, token_usage: None, completed: None, + steer_count: 0, }); turn_state.thread_id = Some(thread_id); turn_state.num_input_images = Some(num_input_images); @@ -250,6 +264,7 @@ impl AnalyticsReducer { started_at: None, token_usage: None, completed: None, + steer_count: 0, }); turn_state.thread_id = Some(input.thread_id); turn_state.token_usage = Some(input.token_usage); @@ -402,6 +417,7 @@ impl AnalyticsReducer { started_at: None, token_usage: None, completed: None, + steer_count: 0, }); turn_state.connection_id = Some(connection_id); turn_state.thread_id = Some(pending_request.thread_id); @@ -427,6 +443,7 @@ impl AnalyticsReducer { started_at: None, token_usage: None, completed: None, + steer_count: 0, }); turn_state.started_at = notification .turn @@ -445,6 +462,7 @@ impl AnalyticsReducer { started_at: None, token_usage: None, completed: None, + steer_count: 0, }); turn_state.completed = Some(CompletedTurnState { status: analytics_turn_status(notification.turn.status), @@ -478,6 +496,8 @@ impl AnalyticsReducer { out: &mut Vec, ) { let thread_source: SessionSource = thread.source.into(); + self.thread_connections + .insert(thread.id.clone(), connection_id); let Some(connection_state) = self.connections.get(&connection_id) else { return; }; @@ -500,6 +520,59 @@ impl AnalyticsReducer { )); } + fn ingest_turn_steer(&mut self, input: TurnSteerInput, out: &mut Vec) { + let TurnSteerInput { + tracking, + turn_steer, + } = input; + if matches!(turn_steer.result, TurnSteerResult::Accepted) + && let Some(accepted_turn_id) = turn_steer.accepted_turn_id.as_ref() + && let Some(turn_state) = self.turns.get_mut(accepted_turn_id) + { + turn_state.steer_count += 1; + } + let Some((app_server_client, runtime)) = + self.connection_metadata_for_turn_steer(&tracking, &turn_steer) + else { + return; + }; + out.push(TrackEventRequest::TurnSteer(CodexTurnSteerEventRequest { + event_type: "codex_turn_steer_event", + event_params: codex_turn_steer_event_params( + app_server_client, + runtime, + &tracking, + turn_steer, + ), + })); + } + + fn connection_metadata_for_turn_steer( + &self, + tracking: &TrackEventsContext, + turn_steer: &CodexTurnSteerEvent, + ) -> Option<(CodexAppServerClientMetadata, CodexRuntimeMetadata)> { + let turn_connection_id = turn_steer + .accepted_turn_id + .as_ref() + .or(turn_steer.expected_turn_id.as_ref()) + .or(Some(&tracking.turn_id)) + .and_then(|turn_id| self.turns.get(turn_id)) + .and_then(|turn_state| turn_state.connection_id); + + let connection_id = turn_connection_id + .or_else(|| self.thread_connections.get(&tracking.thread_id).copied()); + + connection_id + .and_then(|connection_id| self.connections.get(&connection_id)) + .map(|connection_state| { + ( + connection_state.app_server_client.clone(), + connection_state.runtime.clone(), + ) + }) + } + fn maybe_emit_turn_event(&mut self, turn_id: &str, out: &mut Vec) { let Some(turn_state) = self.turns.get(turn_id) else { return; @@ -585,7 +658,7 @@ fn codex_turn_event_params( is_first_turn, status: completed.status, turn_error: completed.turn_error, - steer_count: None, + steer_count: Some(turn_state.steer_count), total_tool_call_count: None, shell_command_count: None, file_change_count: None, diff --git a/codex-rs/app-server/tests/suite/v2/turn_steer.rs b/codex-rs/app-server/tests/suite/v2/turn_steer.rs index 5d1b3cc22e..76da7efb78 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_steer.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_steer.rs @@ -6,6 +6,7 @@ use app_test_support::create_mock_responses_server_sequence; use app_test_support::create_mock_responses_server_sequence_unchecked; use app_test_support::create_shell_command_sse_response; use app_test_support::to_response; +use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url; use codex_app_server::INPUT_TOO_LARGE_ERROR_CODE; use codex_app_server::INVALID_PARAMS_ERROR_CODE; use codex_app_server_protocol::JSONRPCError; @@ -23,6 +24,9 @@ use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use tempfile::TempDir; use tokio::time::timeout; +use super::analytics::enable_analytics_capture; +use super::analytics::wait_for_analytics_event; + const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] @@ -32,7 +36,12 @@ async fn turn_steer_requires_active_turn() -> Result<()> { std::fs::create_dir(&codex_home)?; let server = create_mock_responses_server_sequence(vec![]).await; - create_config_toml(&codex_home, &server.uri())?; + write_mock_responses_config_toml_with_chatgpt_base_url( + &codex_home, + &server.uri(), + &server.uri(), + )?; + enable_analytics_capture(&server, &codex_home).await?; let mut mcp = McpProcess::new(&codex_home).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -52,7 +61,7 @@ async fn turn_steer_requires_active_turn() -> Result<()> { let steer_req = mcp .send_turn_steer_request(TurnSteerParams { - thread_id: thread.id, + thread_id: thread.id.clone(), input: vec![V2UserInput::Text { text: "steer".to_string(), text_elements: Vec::new(), @@ -67,6 +76,21 @@ async fn turn_steer_requires_active_turn() -> Result<()> { .await??; assert_eq!(steer_err.error.code, -32600); + let event = + wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_steer_event").await?; + assert_eq!(event["event_params"]["thread_id"], thread.id); + assert_eq!(event["event_params"]["result"], "rejected"); + assert_eq!(event["event_params"]["num_input_images"], 0); + assert_eq!( + event["event_params"]["expected_turn_id"], + "turn-does-not-exist" + ); + assert_eq!( + event["event_params"]["accepted_turn_id"], + serde_json::Value::Null + ); + assert_eq!(event["event_params"]["rejection_reason"], "no_active_turn"); + Ok(()) } @@ -95,7 +119,12 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> { "call_sleep", )?]) .await; - create_config_toml(&codex_home, &server.uri())?; + write_mock_responses_config_toml_with_chatgpt_base_url( + &codex_home, + &server.uri(), + &server.uri(), + )?; + enable_analytics_capture(&server, &codex_home).await?; let mut mcp = McpProcess::new(&codex_home).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -198,7 +227,12 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> { "call_sleep", )?]) .await; - create_config_toml(&codex_home, &server.uri())?; + write_mock_responses_config_toml_with_chatgpt_base_url( + &codex_home, + &server.uri(), + &server.uri(), + )?; + enable_analytics_capture(&server, &codex_home).await?; let mut mcp = McpProcess::new(&codex_home).await?; timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; @@ -258,31 +292,20 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> { let steer: TurnSteerResponse = to_response::(steer_resp)?; assert_eq!(steer.turn_id, turn.id); + let event = + wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_steer_event").await?; + assert_eq!(event["event_params"]["thread_id"], thread.id); + assert_eq!(event["event_params"]["result"], "accepted"); + assert_eq!(event["event_params"]["num_input_images"], 0); + assert_eq!(event["event_params"]["expected_turn_id"], turn.id); + assert_eq!(event["event_params"]["accepted_turn_id"], turn.id); + assert_eq!( + event["event_params"]["rejection_reason"], + serde_json::Value::Null + ); + mcp.interrupt_turn_and_wait_for_aborted(thread.id, steer.turn_id, DEFAULT_READ_TIMEOUT) .await?; Ok(()) } - -fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { - let config_toml = codex_home.join("config.toml"); - std::fs::write( - config_toml, - format!( - r#" -model = "mock-model" -approval_policy = "never" -sandbox_mode = "danger-full-access" - -model_provider = "mock_provider" - -[model_providers.mock_provider] -name = "Mock provider for test" -base_url = "{server_uri}/v1" -wire_api = "responses" -request_max_retries = 0 -stream_max_retries = 0 -"# - ), - ) -} diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 6dddc247e0..81bc3d7141 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -49,9 +49,13 @@ use chrono::Local; use chrono::Utc; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AppInvocation; +use codex_analytics::CodexTurnSteerEvent; use codex_analytics::InvocationType; use codex_analytics::SubAgentThreadStartedInput; +use codex_analytics::TrackEventsContext; use codex_analytics::TurnResolvedConfigFact; +use codex_analytics::TurnSteerRejectionReason; +use codex_analytics::TurnSteerResult; use codex_analytics::TurnSubmissionType; use codex_analytics::build_track_events_context; use codex_app_server_protocol::McpServerElicitationRequest; @@ -240,6 +244,32 @@ impl SteerInputError { }, } } + + fn to_turn_steer_rejection_reason(&self) -> TurnSteerRejectionReason { + match self { + Self::NoActiveTurn(_) => TurnSteerRejectionReason::NoActiveTurn, + Self::ExpectedTurnMismatch { .. } => TurnSteerRejectionReason::ExpectedTurnMismatch, + Self::ActiveTurnNotSteerable { turn_kind } => match turn_kind { + NonSteerableTurnKind::Review => TurnSteerRejectionReason::NonSteerableReview, + NonSteerableTurnKind::Compact => TurnSteerRejectionReason::NonSteerableCompact, + }, + Self::EmptyInput => TurnSteerRejectionReason::EmptyInput, + } + } +} + +struct AcceptedSteerInput { + turn_id: String, + tracking: TrackEventsContext, + expected_turn_id: String, + num_input_images: usize, +} + +struct RejectedSteerInput { + error: SteerInputError, + tracking: TrackEventsContext, + expected_turn_id: Option, + num_input_images: usize, } /// Notes from the previous real user turn. @@ -4096,47 +4126,163 @@ impl Session { input: Vec, expected_turn_id: Option<&str>, ) -> Result { - if input.is_empty() { - return Err(SteerInputError::EmptyInput); + let created_at = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + match self + .try_append_input_to_active_turn(input, expected_turn_id) + .await + { + Ok(accepted) => { + self.services.analytics_events_client.track_turn_steer( + accepted.tracking, + CodexTurnSteerEvent { + expected_turn_id: Some(accepted.expected_turn_id), + accepted_turn_id: Some(accepted.turn_id.clone()), + num_input_images: accepted.num_input_images, + result: TurnSteerResult::Accepted, + rejection_reason: None, + created_at, + }, + ); + Ok(accepted.turn_id) + } + Err(rejected) => { + self.services.analytics_events_client.track_turn_steer( + rejected.tracking, + CodexTurnSteerEvent { + expected_turn_id: rejected.expected_turn_id, + accepted_turn_id: None, + num_input_images: rejected.num_input_images, + result: TurnSteerResult::Rejected, + rejection_reason: Some(rejected.error.to_turn_steer_rejection_reason()), + created_at, + }, + ); + Err(rejected.error) + } } + } + + async fn try_append_input_to_active_turn( + &self, + input: Vec, + expected_turn_id: Option<&str>, + ) -> Result { + let thread_id = self.conversation_id.to_string(); + let fallback_tracking = || { + build_track_events_context( + String::new(), + thread_id.clone(), + expected_turn_id.unwrap_or_default().to_string(), + ) + }; + + if input.is_empty() { + return Err(RejectedSteerInput { + error: SteerInputError::EmptyInput, + tracking: fallback_tracking(), + expected_turn_id: expected_turn_id.map(str::to_string), + num_input_images: 0, + }); + } + + let num_input_images = input + .iter() + .filter(|item| matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. })) + .count(); let mut active = self.active_turn.lock().await; let Some(active_turn) = active.as_mut() else { - return Err(SteerInputError::NoActiveTurn(input)); + return Err(RejectedSteerInput { + error: SteerInputError::NoActiveTurn(input), + tracking: fallback_tracking(), + expected_turn_id: expected_turn_id.map(str::to_string), + num_input_images, + }); }; - let Some((active_turn_id, _)) = active_turn.tasks.first() else { - return Err(SteerInputError::NoActiveTurn(input)); + let Some((active_turn_id, task)) = active_turn.tasks.first() else { + return Err(RejectedSteerInput { + error: SteerInputError::NoActiveTurn(input), + tracking: fallback_tracking(), + expected_turn_id: expected_turn_id.map(str::to_string), + num_input_images, + }); }; + let active_turn_id = active_turn_id.clone(); + let tracking = build_track_events_context( + task.turn_context.model_info.slug.clone(), + thread_id.clone(), + task.turn_context.sub_id.clone(), + ); if let Some(expected_turn_id) = expected_turn_id && expected_turn_id != active_turn_id { - return Err(SteerInputError::ExpectedTurnMismatch { - expected: expected_turn_id.to_string(), - actual: active_turn_id.clone(), + return Err(RejectedSteerInput { + error: SteerInputError::ExpectedTurnMismatch { + expected: expected_turn_id.to_string(), + actual: active_turn_id, + }, + tracking, + expected_turn_id: Some(expected_turn_id.to_string()), + num_input_images, }); } match active_turn.tasks.first().map(|(_, task)| task.kind) { Some(crate::state::TaskKind::Regular) => {} Some(crate::state::TaskKind::Review) => { - return Err(SteerInputError::ActiveTurnNotSteerable { - turn_kind: NonSteerableTurnKind::Review, + return Err(RejectedSteerInput { + error: SteerInputError::ActiveTurnNotSteerable { + turn_kind: NonSteerableTurnKind::Review, + }, + tracking, + expected_turn_id: expected_turn_id + .map(str::to_string) + .or(Some(active_turn_id)), + num_input_images, }); } Some(crate::state::TaskKind::Compact) => { - return Err(SteerInputError::ActiveTurnNotSteerable { - turn_kind: NonSteerableTurnKind::Compact, + return Err(RejectedSteerInput { + error: SteerInputError::ActiveTurnNotSteerable { + turn_kind: NonSteerableTurnKind::Compact, + }, + tracking, + expected_turn_id: expected_turn_id + .map(str::to_string) + .or(Some(active_turn_id)), + num_input_images, + }); + } + None => { + return Err(RejectedSteerInput { + error: SteerInputError::NoActiveTurn(input), + tracking: fallback_tracking(), + expected_turn_id: expected_turn_id.map(str::to_string), + num_input_images, }); } - None => return Err(SteerInputError::NoActiveTurn(input)), } let mut turn_state = active_turn.turn_state.lock().await; turn_state.push_pending_input(input.into()); turn_state.accept_mailbox_delivery_for_current_turn(); - Ok(active_turn_id.clone()) + drop(turn_state); + + let expected_turn_id = expected_turn_id + .map(str::to_string) + .unwrap_or_else(|| active_turn_id.clone()); + + Ok(AcceptedSteerInput { + turn_id: active_turn_id, + tracking, + expected_turn_id, + num_input_images, + }) } /// Returns the input if there was no task running to inject into. @@ -4801,6 +4947,7 @@ fn submission_dispatch_span(sub: &Submission) -> tracing::Span { /// Operation handlers mod handlers { + use crate::codex::RejectedSteerInput; use crate::codex::Session; use crate::codex::SessionSettingsUpdate; use crate::codex::SteerInputError; @@ -4961,11 +5108,14 @@ mod handlers { sess.maybe_emit_unknown_model_warning_for_turn(current_context.as_ref()) .await; match sess - .steer_input(items.clone(), /*expected_turn_id*/ None) + .try_append_input_to_active_turn(items.clone(), /*expected_turn_id*/ None) .await { Ok(_) => current_context.session_telemetry.user_prompt(&items), - Err(SteerInputError::NoActiveTurn(items)) => { + Err(RejectedSteerInput { + error: SteerInputError::NoActiveTurn(items), + .. + }) => { current_context.session_telemetry.user_prompt(&items); sess.refresh_mcp_servers_if_requested(¤t_context) .await; @@ -4976,10 +5126,10 @@ mod handlers { ) .await; } - Err(err) => { + Err(RejectedSteerInput { error, .. }) => { sess.send_event_raw(Event { id: sub_id, - msg: EventMsg::Error(err.to_error_event()), + msg: EventMsg::Error(error.to_error_event()), }) .await; }