From 8a6be13ae6878cda7b1939ad2a0efc1747bed679 Mon Sep 17 00:00:00 2001 From: Roy Han Date: Tue, 24 Mar 2026 14:03:39 -0700 Subject: [PATCH] define events and support queueing --- codex-rs/core/src/analytics_client.rs | 29 ++++++++++--------- codex-rs/core/src/analytics_client_tests.rs | 10 +++---- codex-rs/core/src/codex.rs | 29 +++++++++---------- codex-rs/core/src/guardian/review_session.rs | 1 + codex-rs/core/tests/suite/items.rs | 11 ++++--- codex-rs/protocol/src/protocol.rs | 5 ++++ codex-rs/tui_app_server/src/app.rs | 2 ++ codex-rs/tui_app_server/src/app_command.rs | 6 ++++ .../tui_app_server/src/app_server_session.rs | 4 ++- codex-rs/tui_app_server/src/chatwidget.rs | 21 ++++++++++++-- 10 files changed, 74 insertions(+), 44 deletions(-) diff --git a/codex-rs/core/src/analytics_client.rs b/codex-rs/core/src/analytics_client.rs index b15179acd3..3c24989001 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -34,6 +34,7 @@ pub(crate) struct TrackEventsContext { #[derive(Clone)] pub(crate) struct TurnMetadata { + pub(crate) turn_type: Option, pub(crate) model_provider: String, pub(crate) sandbox_policy: SandboxPolicy, pub(crate) reasoning_effort: Option, @@ -47,16 +48,16 @@ pub(crate) struct TurnMetadata { pub(crate) num_input_images: usize, } -#[derive(Clone, Copy)] -pub(crate) struct InputMessageMetadata { - pub(crate) message_role: InputMessageRole, - pub(crate) user_message_type: UserMessageType, -} - #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] -pub(crate) enum InputMessageRole { - User, +pub(crate) enum TurnType { + Prompt, + PromptQueued, +} + +#[derive(Clone, Copy)] +pub(crate) struct UserMessageMetadata { + pub(crate) user_message_type: UserMessageType, } pub(crate) fn build_track_events_context( @@ -255,7 +256,7 @@ impl AnalyticsEventsClient { pub(crate) fn track_input_message_metadata( &self, tracking: TrackEventsContext, - input_message_metadata: InputMessageMetadata, + input_message_metadata: UserMessageMetadata, ) { track_input_message_metadata( &self.queue, @@ -342,7 +343,7 @@ struct TrackTurnMetadataJob { struct TrackInputMessageMetadataJob { config: Arc, tracking: TrackEventsContext, - input_message_metadata: InputMessageMetadata, + input_message_metadata: UserMessageMetadata, } struct TrackPluginUsedJob { @@ -435,6 +436,7 @@ struct CodexTurnMetadata { turn_id: String, product_client_id: Option, model: Option, + turn_type: Option, model_provider: String, sandbox_policy: Option<&'static str>, reasoning_effort: Option, @@ -460,7 +462,6 @@ struct CodexInputMessageMetadata { turn_id: Option, product_client_id: Option, model_slug: Option, - message_role: Option, user_message_type: Option, } @@ -595,7 +596,7 @@ pub(crate) fn track_input_message_metadata( queue: &AnalyticsEventsQueue, config: Arc, tracking: Option, - input_message_metadata: InputMessageMetadata, + input_message_metadata: UserMessageMetadata, ) { if config.analytics_enabled == Some(false) { return; @@ -844,6 +845,7 @@ fn codex_turn_metadata( turn_id: tracking.turn_id.clone(), product_client_id: Some(crate::default_client::originator().value), model: Some(tracking.model_slug.clone()), + turn_type: turn_metadata.turn_type, model_provider: turn_metadata.model_provider, sandbox_policy: Some(sandbox_policy_mode(&turn_metadata.sandbox_policy)), reasoning_effort: turn_metadata @@ -865,14 +867,13 @@ fn codex_turn_metadata( fn codex_input_message_metadata( tracking: &TrackEventsContext, - input_message_metadata: InputMessageMetadata, + input_message_metadata: UserMessageMetadata, ) -> CodexInputMessageMetadata { CodexInputMessageMetadata { thread_id: Some(tracking.thread_id.clone()), turn_id: Some(tracking.turn_id.clone()), product_client_id: Some(crate::default_client::originator().value), model_slug: Some(tracking.model_slug.clone()), - message_role: Some(input_message_metadata.message_role), user_message_type: Some(input_message_metadata.user_message_type), } } diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index 5329cac77d..49fabe04a5 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -6,12 +6,12 @@ use super::CodexInputMessageMetadataEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; use super::CodexTurnMetadataEventRequest; -use super::InputMessageMetadata; -use super::InputMessageRole; use super::InvocationType; use super::TrackEventRequest; use super::TrackEventsContext; use super::TurnMetadata; +use super::TurnType; +use super::UserMessageMetadata; use super::UserMessageType; use super::codex_app_metadata; use super::codex_input_message_metadata; @@ -214,6 +214,7 @@ fn turn_metadata_event_serializes_expected_shape() { &tracking, TurnMetadata { model_provider: "openai".to_string(), + turn_type: Some(TurnType::Prompt), sandbox_policy: SandboxPolicy::new_read_only_policy(), reasoning_effort: Some(ReasoningEffort::High), reasoning_summary: Some(ReasoningSummary::Detailed), @@ -239,6 +240,7 @@ fn turn_metadata_event_serializes_expected_shape() { "turn_id": "turn-2", "product_client_id": crate::default_client::originator().value, "model": "gpt-5", + "turn_type": "prompt", "model_provider": "openai", "sandbox_policy": "read_only", "reasoning_effort": "high", @@ -266,8 +268,7 @@ fn input_message_metadata_event_serializes_expected_shape() { event_type: "codex_input_message_metadata", event_params: codex_input_message_metadata( &tracking, - InputMessageMetadata { - message_role: InputMessageRole::User, + UserMessageMetadata { user_message_type: UserMessageType::PromptSteering, }, ), @@ -284,7 +285,6 @@ fn input_message_metadata_event_serializes_expected_shape() { "turn_id": "turn-2", "product_client_id": crate::default_client::originator().value, "model_slug": "gpt-5", - "message_role": "user", "user_message_type": "prompt_steering" } }) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index e60ea1d581..41d90e1fac 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -14,11 +14,10 @@ use crate::agent::AgentStatus; use crate::agent::agent_status_from_event; use crate::analytics_client::AnalyticsEventsClient; use crate::analytics_client::AppInvocation; -use crate::analytics_client::InputMessageMetadata; -use crate::analytics_client::InputMessageRole; use crate::analytics_client::InvocationType; use crate::analytics_client::TurnMetadata; -use crate::analytics_client::UserMessageType; +use crate::analytics_client::TurnType; +use crate::analytics_client::UserMessageMetadata; use crate::analytics_client::build_track_events_context; use crate::apps::render_apps_section; use crate::auth_env_telemetry::collect_auth_env_telemetry; @@ -949,6 +948,7 @@ impl TurnContext { provider: self.provider.clone(), reasoning_effort, reasoning_summary: self.reasoning_summary, + user_message_type: None, session_source: self.session_source.clone(), environment: Arc::clone(&self.environment), cwd: self.cwd.clone(), @@ -3942,9 +3942,8 @@ impl Session { .analytics_events_client .track_input_message_metadata( tracking, - InputMessageMetadata { - message_role: InputMessageRole::User, - user_message_type: UserMessageType::PromptSteering, + UserMessageMetadata { + user_message_type: SubmittedUserMessageType::PromptSteering, }, ); } @@ -4596,6 +4595,7 @@ mod handlers { items, collaboration_mode, personality, + user_message_type, } => { let collaboration_mode = collaboration_mode.or_else(|| { Some(CollaborationMode { @@ -4621,7 +4621,7 @@ mod handlers { final_output_json_schema: Some(final_output_json_schema), personality, app_server_client_name: None, - user_message_type: None, + user_message_type, }, ) } @@ -5412,6 +5412,7 @@ async fn spawn_review_thread( provider: provider_for_context, reasoning_effort, reasoning_summary, + user_message_type: None, session_source, environment: Arc::clone(&parent_turn_context.environment), tools_config, @@ -5714,19 +5715,15 @@ pub(crate) async fn run_turn( let user_message_type = turn_context .user_message_type .unwrap_or(SubmittedUserMessageType::Prompt); - sess.services - .analytics_events_client - .track_input_message_metadata( - tracking.clone(), - InputMessageMetadata { - message_role: InputMessageRole::User, - user_message_type, - }, - ); sess.services.analytics_events_client.track_turn_metadata( tracking.clone(), TurnMetadata { model_provider: turn_context.config.model_provider_id.clone(), + turn_type: match user_message_type { + SubmittedUserMessageType::Prompt => Some(TurnType::Prompt), + SubmittedUserMessageType::PromptQueued => Some(TurnType::PromptQueued), + SubmittedUserMessageType::PromptSteering => None, + }, sandbox_policy: turn_context.sandbox_policy.get().clone(), reasoning_effort: turn_context.reasoning_effort, reasoning_summary: Some(turn_context.reasoning_summary), diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 50bf2ed843..a04242050e 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -523,6 +523,7 @@ async fn run_review_on_session( final_output_json_schema: Some(params.schema.clone()), collaboration_mode: None, personality: params.personality, + user_message_type: None, }) .await }), diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index 855b14dcf0..5b7ba35773 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -242,6 +242,7 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> { let event_params = &event["event_params"]; + assert_eq!(event_params["turn_type"], "prompt"); assert_eq!(event_params["sandbox_policy"], "read_only"); assert_eq!( event_params["product_client_id"], @@ -271,7 +272,7 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn user_turn_tracks_input_message_prompt_metadata_analytics() -> anyhow::Result<()> { +async fn user_turn_tracks_turn_type_prompt_metadata_analytics() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; @@ -323,14 +324,13 @@ async fn user_turn_tracks_input_message_prompt_metadata_analytics() -> anyhow::R wait_for_event(&codex, |event| matches!(event, EventMsg::TurnComplete(_))).await; - let event = wait_for_analytics_event(&server, "codex_input_message_metadata", |event| { - event["event_params"]["user_message_type"] == "prompt" + let event = wait_for_analytics_event(&server, "codex_turn_metadata", |event| { + event["event_params"]["turn_type"] == "prompt" }) .await; let event_params = &event["event_params"]; - assert_eq!(event_params["message_role"], "user"); - assert_eq!(event_params["user_message_type"], "prompt"); + assert_eq!(event_params["turn_type"], "prompt"); assert_eq!( event_params["product_client_id"], serde_json::json!(codex_core::default_client::originator().value) @@ -444,7 +444,6 @@ async fn user_turn_tracks_input_message_prompt_steering_metadata_analytics() -> .await; let event_params = &event["event_params"]; - assert_eq!(event_params["message_role"], "user"); assert_eq!(event_params["user_message_type"], "prompt_steering"); assert_eq!( event_params["product_client_id"], diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 284b78674b..a0ede9804f 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -319,6 +319,11 @@ pub enum Op { /// Optional personality override for this turn. #[serde(skip_serializing_if = "Option::is_none")] personality: Option, + + /// Semantic classification of this submitted user message when known by + /// the higher-level surface. + #[serde(default, skip_serializing_if = "Option::is_none")] + user_message_type: Option, }, /// Inter-agent communication that should be recorded as assistant history diff --git a/codex-rs/tui_app_server/src/app.rs b/codex-rs/tui_app_server/src/app.rs index 29da170b82..ded6b151f5 100644 --- a/codex-rs/tui_app_server/src/app.rs +++ b/codex-rs/tui_app_server/src/app.rs @@ -2020,6 +2020,7 @@ impl App { final_output_json_schema, collaboration_mode, personality, + user_message_type, } => { if let Some(turn_id) = self.active_turn_id_for_thread(thread_id).await { match app_server @@ -2053,6 +2054,7 @@ impl App { *service_tier, collaboration_mode.clone(), *personality, + *user_message_type, final_output_json_schema.clone(), ) .await?; diff --git a/codex-rs/tui_app_server/src/app_command.rs b/codex-rs/tui_app_server/src/app_command.rs index e01a250274..658c8f5a73 100644 --- a/codex-rs/tui_app_server/src/app_command.rs +++ b/codex-rs/tui_app_server/src/app_command.rs @@ -17,6 +17,7 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::ReviewDecision; use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::UserMessageType; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_protocol::user_input::UserInput; @@ -51,6 +52,7 @@ pub(crate) enum AppCommandView<'a> { final_output_json_schema: &'a Option, collaboration_mode: &'a Option, personality: &'a Option, + user_message_type: &'a Option, }, OverrideTurnContext { cwd: &'a Option, @@ -155,6 +157,7 @@ impl AppCommand { final_output_json_schema: Option, collaboration_mode: Option, personality: Option, + user_message_type: Option, ) -> Self { Self(Op::UserTurn { items, @@ -169,6 +172,7 @@ impl AppCommand { final_output_json_schema, collaboration_mode, personality, + user_message_type, }) } @@ -314,6 +318,7 @@ impl AppCommand { final_output_json_schema, collaboration_mode, personality, + user_message_type, } => AppCommandView::UserTurn { items, cwd, @@ -327,6 +332,7 @@ impl AppCommand { final_output_json_schema, collaboration_mode, personality, + user_message_type, }, Op::OverrideTurnContext { cwd, diff --git a/codex-rs/tui_app_server/src/app_server_session.rs b/codex-rs/tui_app_server/src/app_server_session.rs index 8274bda7a2..e70ae275a5 100644 --- a/codex-rs/tui_app_server/src/app_server_session.rs +++ b/codex-rs/tui_app_server/src/app_server_session.rs @@ -75,6 +75,7 @@ use codex_protocol::protocol::ReviewRequest; use codex_protocol::protocol::ReviewTarget as CoreReviewTarget; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionNetworkProxyRuntime; +use codex_protocol::protocol::UserMessageType; use color_eyre::eyre::ContextCompat; use color_eyre::eyre::Result; use color_eyre::eyre::WrapErr; @@ -389,6 +390,7 @@ impl AppServerSession { service_tier: Option>, collaboration_mode: Option, personality: Option, + user_message_type: Option, output_schema: Option, ) -> Result { let request_id = self.next_request_id(); @@ -407,7 +409,7 @@ impl AppServerSession { effort, summary, personality, - user_message_type: None, + user_message_type, output_schema, collaboration_mode, }, diff --git a/codex-rs/tui_app_server/src/chatwidget.rs b/codex-rs/tui_app_server/src/chatwidget.rs index b444dc13bc..9907ac4e01 100644 --- a/codex-rs/tui_app_server/src/chatwidget.rs +++ b/codex-rs/tui_app_server/src/chatwidget.rs @@ -2706,7 +2706,10 @@ impl ChatWidget { .map(|pending| pending.user_message) .collect(); if !pending_steers.is_empty() { - self.submit_user_message(merge_user_messages(pending_steers)); + self.submit_user_message_with_type( + merge_user_messages(pending_steers), + Some(UserMessageType::PromptSteering), + ); } else if let Some(combined) = self.drain_pending_messages_for_restore() { self.restore_user_message_to_composer(combined); } @@ -5210,6 +5213,14 @@ impl ChatWidget { } fn submit_user_message(&mut self, user_message: UserMessage) { + self.submit_user_message_with_type(user_message, None); + } + + fn submit_user_message_with_type( + &mut self, + user_message: UserMessage, + user_message_type: Option, + ) { if !self.is_session_configured() { tracing::warn!("cannot submit user message before session is configured; queueing"); self.queued_user_messages.push_front(user_message); @@ -5425,6 +5436,7 @@ impl ChatWidget { /*final_output_json_schema*/ None, collaboration_mode, personality, + user_message_type, ); if !self.submit_op(op) { @@ -6884,8 +6896,13 @@ impl ChatWidget { if self.bottom_pane.is_task_running() { return; } + let user_message_type = if self.rejected_steers_queue.is_empty() { + UserMessageType::PromptQueued + } else { + UserMessageType::PromptSteering + }; if let Some(user_message) = self.pop_next_queued_user_message() { - self.submit_user_message(user_message); + self.submit_user_message_with_type(user_message, Some(user_message_type)); } // Update the list to reflect the remaining queued messages (if any). self.refresh_pending_input_preview();