From 69a755c9f61092ce261345c945fb018084706eda Mon Sep 17 00:00:00 2001 From: Roy Han Date: Tue, 24 Mar 2026 11:12:37 -0700 Subject: [PATCH] user message metadata --- .../app-server-protocol/src/protocol/v2.rs | 8 +- codex-rs/app-server-test-client/src/lib.rs | 6 + .../app-server/src/codex_message_processor.rs | 3 +- .../src/message_processor/tracing_tests.rs | 1 + codex-rs/app-server/tests/suite/v2/account.rs | 4 + .../app-server/tests/suite/v2/compaction.rs | 1 + .../v2/connection_handling_websocket_unix.rs | 1 + .../tests/suite/v2/dynamic_tools.rs | 4 + .../app-server/tests/suite/v2/initialize.rs | 1 + .../tests/suite/v2/mcp_server_elicitation.rs | 2 + .../tests/suite/v2/output_schema.rs | 3 + .../app-server/tests/suite/v2/plan_item.rs | 1 + .../tests/suite/v2/request_permissions.rs | 1 + .../tests/suite/v2/request_user_input.rs | 1 + codex-rs/app-server/tests/suite/v2/review.rs | 1 + .../tests/suite/v2/safety_check_downgrade.rs | 2 + .../tests/suite/v2/thread_archive.rs | 3 + .../app-server/tests/suite/v2/thread_fork.rs | 1 + .../app-server/tests/suite/v2/thread_list.rs | 2 + .../app-server/tests/suite/v2/thread_read.rs | 1 + .../tests/suite/v2/thread_resume.rs | 18 ++ .../tests/suite/v2/thread_rollback.rs | 2 + .../tests/suite/v2/thread_shell_command.rs | 1 + .../tests/suite/v2/thread_status.rs | 2 + .../tests/suite/v2/thread_unarchive.rs | 1 + .../tests/suite/v2/thread_unsubscribe.rs | 2 + .../tests/suite/v2/turn_interrupt.rs | 2 + .../app-server/tests/suite/v2/turn_start.rs | 25 ++ .../tests/suite/v2/turn_start_zsh_fork.rs | 4 + .../app-server/tests/suite/v2/turn_steer.rs | 2 + codex-rs/core/src/analytics_client.rs | 106 +++++++++ codex-rs/core/src/analytics_client_tests.rs | 41 ++++ codex-rs/core/src/codex.rs | 65 +++++- codex-rs/core/tests/suite/items.rs | 220 ++++++++++++++++++ codex-rs/debug-client/src/client.rs | 1 + codex-rs/exec/src/lib.rs | 1 + codex-rs/protocol/src/protocol.rs | 22 ++ .../tui_app_server/src/app_server_session.rs | 1 + 38 files changed, 559 insertions(+), 4 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 3e30f81323..52434a9c68 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -84,6 +84,7 @@ use codex_protocol::protocol::SkillToolDependency as CoreSkillToolDependency; use codex_protocol::protocol::SubAgentSource as CoreSubAgentSource; use codex_protocol::protocol::TokenUsage as CoreTokenUsage; use codex_protocol::protocol::TokenUsageInfo as CoreTokenUsageInfo; +use codex_protocol::protocol::UserMessageType; use codex_protocol::request_permissions::PermissionGrantScope as CorePermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile; use codex_protocol::user_input::ByteRange as CoreByteRange; @@ -3915,6 +3916,10 @@ pub struct TurnStartParams { /// Override the personality for this turn and subsequent turns. #[ts(optional = nullable)] pub personality: Option, + /// Classify the primary input message submitted with `turn/start`. + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional = nullable)] + pub user_message_type: Option, /// Optional JSON Schema used to constrain the final assistant message for /// this turn. #[ts(optional = nullable)] @@ -8034,9 +8039,10 @@ mod tests { service_tier: None, effort: None, summary: None, + personality: None, + user_message_type: None, output_schema: None, collaboration_mode: None, - personality: None, }; let serialized_without_override = serde_json::to_value(&without_override).expect("params should serialize"); diff --git a/codex-rs/app-server-test-client/src/lib.rs b/codex-rs/app-server-test-client/src/lib.rs index a479f96529..53ba89a133 100644 --- a/codex-rs/app-server-test-client/src/lib.rs +++ b/codex-rs/app-server-test-client/src/lib.rs @@ -735,6 +735,7 @@ async fn trigger_zsh_fork_multi_cmd_approval( text: message, text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }; turn_params.approval_policy = Some(AskForApproval::OnRequest); @@ -820,6 +821,7 @@ async fn resume_message_v2( text: user_message, text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() })?; println!("< turn/start response: {turn_response:?}"); @@ -964,6 +966,7 @@ async fn send_message_v2_with_policies( // Test client sends plain text without UI element ranges. text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }; turn_params.approval_policy = policies.approval_policy; @@ -1004,6 +1007,7 @@ async fn send_follow_up_v2( // Test client sends plain text without UI element ranges. text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }; let first_turn_response = client.turn_start(first_turn_params)?; @@ -1017,6 +1021,7 @@ async fn send_follow_up_v2( // Test client sends plain text without UI element ranges. text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }; let follow_up_response = client.turn_start(follow_up_params)?; @@ -1240,6 +1245,7 @@ fn live_elicitation_timeout_pause( sandbox_policy: Some(SandboxPolicy::DangerFullAccess), effort: Some(ReasoningEffort::High), cwd: Some(workspace), + user_message_type: None, ..Default::default() })?; println!("< turn/start response: {turn_response:?}"); diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 947a49ff2f..aae9fa49e3 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6069,9 +6069,10 @@ impl CodexMessageProcessor { .submit_core_op( &request_id, thread.as_ref(), - Op::UserInput { + Op::UserInputWithMetadata { items: mapped_items, final_output_json_schema: params.output_schema, + user_message_type: params.user_message_type, }, ) .await; diff --git a/codex-rs/app-server/src/message_processor/tracing_tests.rs b/codex-rs/app-server/src/message_processor/tracing_tests.rs index 58499745bd..588fd8c422 100644 --- a/codex-rs/app-server/src/message_processor/tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor/tracing_tests.rs @@ -599,6 +599,7 @@ async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> { effort: None, summary: None, personality: None, + user_message_type: None, output_schema: None, collaboration_mode: None, }, diff --git a/codex-rs/app-server/tests/suite/v2/account.rs b/codex-rs/app-server/tests/suite/v2/account.rs index 514a6542a0..acacb4df84 100644 --- a/codex-rs/app-server/tests/suite/v2/account.rs +++ b/codex-rs/app-server/tests/suite/v2/account.rs @@ -419,6 +419,7 @@ async fn external_auth_refreshes_on_unauthorized() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -525,6 +526,7 @@ async fn external_auth_refresh_error_fails_turn() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -647,6 +649,7 @@ async fn external_auth_refresh_mismatched_workspace_fails_turn() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -762,6 +765,7 @@ async fn external_auth_refresh_invalid_access_token_fails_turn() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/compaction.rs b/codex-rs/app-server/tests/suite/v2/compaction.rs index c0922d3256..70c3a3e3c9 100644 --- a/codex-rs/app-server/tests/suite/v2/compaction.rs +++ b/codex-rs/app-server/tests/suite/v2/compaction.rs @@ -342,6 +342,7 @@ async fn send_turn_and_wait(mcp: &mut McpProcess, thread_id: &str, text: &str) - text: text.to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs index 08156b9d95..83c90f1492 100644 --- a/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs +++ b/codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs @@ -202,6 +202,7 @@ async fn send_turn_start_request(stream: &mut WsClient, id: i64, thread_id: &str text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() })?), ) diff --git a/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs b/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs index 0ab3f47235..2ec81835fc 100644 --- a/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs +++ b/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs @@ -86,6 +86,7 @@ async fn thread_start_injects_dynamic_tools_into_model_requests() -> Result<()> text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -164,6 +165,7 @@ async fn thread_start_keeps_hidden_dynamic_tools_out_of_model_requests() -> Resu text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -252,6 +254,7 @@ async fn dynamic_tool_call_round_trip_sends_text_content_items_to_model() -> Res text: "Run the tool".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -420,6 +423,7 @@ async fn dynamic_tool_call_round_trip_sends_content_items_to_model() -> Result<( text: "Run the tool".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/initialize.rs b/codex-rs/app-server/tests/suite/v2/initialize.rs index 9b5f0cacc8..d22917391f 100644 --- a/codex-rs/app-server/tests/suite/v2/initialize.rs +++ b/codex-rs/app-server/tests/suite/v2/initialize.rs @@ -248,6 +248,7 @@ async fn turn_start_notify_payload_includes_initialize_client_name() -> Result<( text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs b/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs index a784b34908..768962cb2f 100644 --- a/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs +++ b/codex-rs/app-server/tests/suite/v2/mcp_server_elicitation.rs @@ -137,6 +137,7 @@ async fn mcp_server_elicitation_round_trip() -> Result<()> { text_elements: Vec::new(), }], model: Some("mock-model".to_string()), + user_message_type: None, ..Default::default() }) .await?; @@ -169,6 +170,7 @@ async fn mcp_server_elicitation_round_trip() -> Result<()> { text_elements: Vec::new(), }], model: Some("mock-model".to_string()), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/output_schema.rs b/codex-rs/app-server/tests/suite/v2/output_schema.rs index 149e098b68..ea0924ebb5 100644 --- a/codex-rs/app-server/tests/suite/v2/output_schema.rs +++ b/codex-rs/app-server/tests/suite/v2/output_schema.rs @@ -63,6 +63,7 @@ async fn turn_start_accepts_output_schema_v2() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, output_schema: Some(output_schema.clone()), ..Default::default() }) @@ -145,6 +146,7 @@ async fn turn_start_output_schema_is_per_turn_v2() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, output_schema: Some(output_schema.clone()), ..Default::default() }) @@ -187,6 +189,7 @@ async fn turn_start_output_schema_is_per_turn_v2() -> Result<()> { text: "Hello again".to_string(), text_elements: Vec::new(), }], + user_message_type: None, output_schema: None, ..Default::default() }) diff --git a/codex-rs/app-server/tests/suite/v2/plan_item.rs b/codex-rs/app-server/tests/suite/v2/plan_item.rs index 0ed93cbeae..b6bdf933b9 100644 --- a/codex-rs/app-server/tests/suite/v2/plan_item.rs +++ b/codex-rs/app-server/tests/suite/v2/plan_item.rs @@ -156,6 +156,7 @@ async fn start_plan_mode_turn(mcp: &mut McpProcess) -> Result Result<()> { text_elements: Vec::new(), }], model: Some("mock-model".to_string()), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/request_user_input.rs b/codex-rs/app-server/tests/suite/v2/request_user_input.rs index f77ddfb4f7..f1ffd70dc6 100644 --- a/codex-rs/app-server/tests/suite/v2/request_user_input.rs +++ b/codex-rs/app-server/tests/suite/v2/request_user_input.rs @@ -57,6 +57,7 @@ async fn request_user_input_round_trip() -> Result<()> { }], model: Some("mock-model".to_string()), effort: Some(ReasoningEffort::Medium), + user_message_type: None, collaboration_mode: Some(CollaborationMode { mode: ModeKind::Plan, settings: Settings { diff --git a/codex-rs/app-server/tests/suite/v2/review.rs b/codex-rs/app-server/tests/suite/v2/review.rs index 6febbaf52a..a5d6275b3a 100644 --- a/codex-rs/app-server/tests/suite/v2/review.rs +++ b/codex-rs/app-server/tests/suite/v2/review.rs @@ -434,6 +434,7 @@ async fn materialize_thread_rollout(mcp: &mut McpProcess, thread_id: &str) -> Re text: "materialize rollout".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs b/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs index 8405d11490..e1fe6faf3d 100644 --- a/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs +++ b/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs @@ -63,6 +63,7 @@ async fn openai_model_header_mismatch_emits_model_rerouted_notification_v2() -> text: "trigger safeguard".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -136,6 +137,7 @@ async fn response_model_field_mismatch_emits_model_rerouted_notification_v2_when text: "trigger response model check".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index 20fd6fd884..b9ef112d3f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -92,6 +92,7 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { text: "materialize".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -189,6 +190,7 @@ async fn thread_archive_clears_stale_subscriptions_before_resume() -> Result<()> text: "materialize".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -266,6 +268,7 @@ async fn thread_archive_clears_stale_subscriptions_before_resume() -> Result<()> text: "secondary turn".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 1b40fadc67..6ed2c07b01 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -463,6 +463,7 @@ async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<() text: "continue".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 75bffe622c..7d2a008619 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -216,6 +216,7 @@ async fn thread_list_reports_system_error_idle_flag_after_failed_turn() -> Resul text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -238,6 +239,7 @@ async fn thread_list_reports_system_error_idle_flag_after_failed_turn() -> Resul text: "fail turn".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index dcc411538e..ef69424e9e 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -441,6 +441,7 @@ async fn thread_read_reports_system_error_idle_flag_after_failed_turn() -> Resul text: "fail this turn".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 4443abd6ec..f5667a54f8 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -564,6 +564,7 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() - text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -613,6 +614,7 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -638,6 +640,7 @@ async fn thread_resume_keeps_in_flight_turn_streaming() -> Result<()> { text: "respond with docs".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -720,6 +723,7 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> { text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -743,6 +747,7 @@ async fn thread_resume_rejects_history_when_thread_is_running() -> Result<()> { text: "keep running".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -836,6 +841,7 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -859,6 +865,7 @@ async fn thread_resume_rejects_mismatched_path_when_thread_is_running() -> Resul text: "keep running".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -942,6 +949,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -964,6 +972,7 @@ async fn thread_resume_rejoins_running_thread_even_with_override_mismatch() -> R text: "keep running".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1055,6 +1064,7 @@ async fn thread_resume_replays_pending_command_execution_request_approval() -> R text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1078,6 +1088,7 @@ async fn thread_resume_replays_pending_command_execution_request_approval() -> R text_elements: Vec::new(), }], approval_policy: Some(AskForApproval::UnlessTrusted), + user_message_type: None, ..Default::default() }) .await?; @@ -1194,6 +1205,7 @@ async fn thread_resume_replays_pending_file_change_request_approval() -> Result< text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -1218,6 +1230,7 @@ async fn thread_resume_replays_pending_file_change_request_approval() -> Result< }], cwd: Some(workspace.clone()), approval_policy: Some(AskForApproval::UnlessTrusted), + user_message_type: None, ..Default::default() }) .await?; @@ -1362,6 +1375,7 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1541,6 +1555,7 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> { text: "materialize".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1661,6 +1676,7 @@ async fn start_materialized_thread_and_restart( text: seed_text.to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1735,6 +1751,7 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { text: "seed history".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1775,6 +1792,7 @@ async fn thread_resume_accepts_personality_override() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs index 3487b9e36a..404469956f 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_rollback.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_rollback.rs @@ -61,6 +61,7 @@ async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<() text: first_text.to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -82,6 +83,7 @@ async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<() text: "Second".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs index 5b58796bf5..102521bebf 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_shell_command.rs @@ -205,6 +205,7 @@ async fn thread_shell_command_uses_existing_active_turn() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_status.rs b/codex-rs/app-server/tests/suite/v2/thread_status.rs index ad90e4900a..25a6bcf709 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_status.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_status.rs @@ -53,6 +53,7 @@ async fn thread_status_changed_emits_runtime_updates() -> Result<()> { text_elements: Vec::new(), }], model: Some("mock-model".to_string()), + user_message_type: None, ..Default::default() }) .await?; @@ -175,6 +176,7 @@ async fn thread_status_changed_can_be_opted_out() -> Result<()> { text_elements: Vec::new(), }], model: Some("mock-model".to_string()), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index b2ae60ae35..f57ac86983 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -60,6 +60,7 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result text: "materialize".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs index 6427ae7ce0..999fda76e4 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unsubscribe.rs @@ -177,6 +177,7 @@ async fn thread_unsubscribe_during_turn_interrupts_turn_and_emits_thread_closed( text_elements: Vec::new(), }], cwd: Some(working_directory), + user_message_type: None, ..Default::default() }) .await?; @@ -250,6 +251,7 @@ async fn thread_unsubscribe_clears_cached_status_before_resume() -> Result<()> { text: "fail this turn".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs index 8f944dca07..a22b1882b0 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_interrupt.rs @@ -78,6 +78,7 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(working_directory.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -172,6 +173,7 @@ async fn turn_interrupt_resolves_pending_command_approval_request() -> Result<() text_elements: Vec::new(), }], cwd: Some(working_directory), + user_message_type: None, ..Default::default() }) .await?; 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 8d7ca02613..7623172808 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start.rs @@ -121,6 +121,7 @@ async fn turn_start_sends_originator_header() -> Result<()> { text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -192,6 +193,7 @@ async fn turn_start_emits_user_message_item_with_text_elements() -> Result<()> { text: "Hello".to_string(), text_elements: text_elements.clone(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -280,6 +282,7 @@ async fn turn_start_accepts_text_at_limit_with_mention_item() -> Result<()> { path: "app://demo-app".to_string(), }, ], + user_message_type: None, ..Default::default() }) .await?; @@ -343,6 +346,7 @@ async fn turn_start_rejects_combined_oversized_text_input() -> Result<()> { text_elements: Vec::new(), }, ], + user_message_type: None, ..Default::default() }) .await?; @@ -419,6 +423,7 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<( text: "Hello".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -468,6 +473,7 @@ async fn turn_start_emits_notifications_and_accepts_model_override() -> Result<( text_elements: Vec::new(), }], model: Some("mock-model-override".to_string()), + user_message_type: None, ..Default::default() }) .await?; @@ -564,6 +570,7 @@ async fn turn_start_accepts_collaboration_mode_override_v2() -> Result<()> { model: Some("mock-model-override".to_string()), effort: Some(ReasoningEffort::Low), summary: Some(ReasoningSummary::Auto), + user_message_type: None, output_schema: None, collaboration_mode: Some(collaboration_mode), ..Default::default() @@ -651,6 +658,7 @@ async fn turn_start_uses_thread_feature_overrides_for_collaboration_mode_instruc model: Some("mock-model-override".to_string()), effort: Some(ReasoningEffort::Low), summary: Some(ReasoningSummary::Auto), + user_message_type: None, output_schema: None, collaboration_mode: Some(collaboration_mode), ..Default::default() @@ -720,6 +728,7 @@ async fn turn_start_accepts_personality_override_v2() -> Result<()> { text_elements: Vec::new(), }], personality: Some(Personality::Friendly), + user_message_type: None, ..Default::default() }) .await?; @@ -801,6 +810,7 @@ async fn turn_start_change_personality_mid_thread_v2() -> Result<()> { text_elements: Vec::new(), }], personality: None, + user_message_type: None, ..Default::default() }) .await?; @@ -825,6 +835,7 @@ async fn turn_start_change_personality_mid_thread_v2() -> Result<()> { text_elements: Vec::new(), }], personality: Some(Personality::Friendly), + user_message_type: None, ..Default::default() }) .await?; @@ -927,6 +938,7 @@ async fn turn_start_uses_migrated_pragmatic_personality_without_override_v2() -> text_elements: Vec::new(), }], personality: None, + user_message_type: None, ..Default::default() }) .await?; @@ -995,6 +1007,7 @@ async fn turn_start_accepts_local_image_input() -> Result<()> { .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), input: vec![V2UserInput::LocalImage { path: image_path }], + user_message_type: None, ..Default::default() }) .await?; @@ -1082,6 +1095,7 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> { text: "run python".to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1151,6 +1165,7 @@ async fn turn_start_exec_approval_toggle_v2() -> Result<()> { model: Some("mock-model".to_string()), effort: Some(ReasoningEffort::Medium), summary: Some(ReasoningSummary::Auto), + user_message_type: None, ..Default::default() }) .await?; @@ -1224,6 +1239,7 @@ async fn turn_start_exec_approval_decline_v2() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -1392,6 +1408,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { summary: Some(ReasoningSummary::Auto), service_tier: None, personality: None, + user_message_type: None, output_schema: None, collaboration_mode: None, }) @@ -1425,6 +1442,7 @@ async fn turn_start_updates_sandbox_and_cwd_between_turns_v2() -> Result<()> { summary: Some(ReasoningSummary::Auto), service_tier: None, personality: None, + user_message_type: None, output_schema: None, collaboration_mode: None, }) @@ -1527,6 +1545,7 @@ async fn turn_start_file_change_approval_v2() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -1744,6 +1763,7 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<() text: PARENT_PROMPT.to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -1958,6 +1978,7 @@ config_file = "./custom-role.toml" text: PARENT_PROMPT.to_string(), text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }) .await?; @@ -2104,6 +2125,7 @@ async fn turn_start_file_change_approval_accept_for_session_persists_v2() -> Res text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -2181,6 +2203,7 @@ async fn turn_start_file_change_approval_accept_for_session_persists_v2() -> Res text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -2284,6 +2307,7 @@ async fn turn_start_file_change_approval_decline_v2() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -2429,6 +2453,7 @@ async fn command_execution_notifications_include_process_id() -> Result<()> { text_elements: Vec::new(), }], sandbox_policy: Some(codex_app_server_protocol::SandboxPolicy::DangerFullAccess), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs b/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs index c8ae882e23..db25549b25 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_start_zsh_fork.rs @@ -127,6 +127,7 @@ async fn turn_start_shell_zsh_fork_executes_command_v2() -> Result<()> { model: Some("mock-model".to_string()), effort: Some(codex_protocol::openai_models::ReasoningEffort::Medium), summary: Some(codex_protocol::config_types::ReasoningSummary::Auto), + user_message_type: None, ..Default::default() }) .await?; @@ -241,6 +242,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_decline_v2() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -374,6 +376,7 @@ async fn turn_start_shell_zsh_fork_exec_approval_cancel_v2() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(workspace.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -544,6 +547,7 @@ async fn turn_start_shell_zsh_fork_subcommand_decline_marks_parent_declined_v2() model: Some("mock-model".to_string()), effort: Some(codex_protocol::openai_models::ReasoningEffort::Medium), summary: Some(codex_protocol::config_types::ReasoningSummary::Auto), + user_message_type: None, ..Default::default() }) .await?; 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..569e547de9 100644 --- a/codex-rs/app-server/tests/suite/v2/turn_steer.rs +++ b/codex-rs/app-server/tests/suite/v2/turn_steer.rs @@ -121,6 +121,7 @@ async fn turn_steer_rejects_oversized_text_input() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(working_directory.clone()), + user_message_type: None, ..Default::default() }) .await?; @@ -224,6 +225,7 @@ async fn turn_steer_returns_active_turn_id() -> Result<()> { text_elements: Vec::new(), }], cwd: Some(working_directory.clone()), + user_message_type: None, ..Default::default() }) .await?; diff --git a/codex-rs/core/src/analytics_client.rs b/codex-rs/core/src/analytics_client.rs index d52cdc6fb9..8789fcdb11 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -10,6 +10,7 @@ use codex_protocol::config_types::ServiceTier; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SkillScope; +use codex_protocol::protocol::UserMessageType; use serde::Serialize; use sha1::Digest; use sha1::Sha1; @@ -37,6 +38,18 @@ pub(crate) struct TurnMetadata { pub(crate) collaboration_mode: ModeKind, } +#[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) fn build_track_events_context( model_slug: String, thread_id: String, @@ -101,6 +114,9 @@ impl AnalyticsEventsQueue { TrackEventsJob::TurnMetadata(job) => { send_track_turn_metadata(&auth_manager, job).await; } + TrackEventsJob::InputMessageMetadata(job) => { + send_track_input_message_metadata(&auth_manager, job).await; + } TrackEventsJob::PluginUsed(job) => { send_track_plugin_used(&auth_manager, job).await; } @@ -227,6 +243,19 @@ impl AnalyticsEventsClient { ); } + pub(crate) fn track_input_message_metadata( + &self, + tracking: TrackEventsContext, + input_message_metadata: InputMessageMetadata, + ) { + track_input_message_metadata( + &self.queue, + Arc::clone(&self.config), + Some(tracking), + input_message_metadata, + ); + } + pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { track_plugin_management( &self.queue, @@ -269,6 +298,7 @@ enum TrackEventsJob { AppMentioned(TrackAppMentionedJob), AppUsed(TrackAppUsedJob), TurnMetadata(TrackTurnMetadataJob), + InputMessageMetadata(TrackInputMessageMetadataJob), PluginUsed(TrackPluginUsedJob), PluginInstalled(TrackPluginManagementJob), PluginUninstalled(TrackPluginManagementJob), @@ -300,6 +330,12 @@ struct TrackTurnMetadataJob { turn_metadata: TurnMetadata, } +struct TrackInputMessageMetadataJob { + config: Arc, + tracking: TrackEventsContext, + input_message_metadata: InputMessageMetadata, +} + struct TrackPluginUsedJob { config: Arc, tracking: TrackEventsContext, @@ -335,6 +371,7 @@ enum TrackEventRequest { AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), TurnMetadata(CodexTurnMetadataEventRequest), + InputMessageMetadata(CodexInputMessageMetadataEventRequest), PluginUsed(CodexPluginUsedEventRequest), PluginInstalled(CodexPluginEventRequest), PluginUninstalled(CodexPluginEventRequest), @@ -402,6 +439,22 @@ struct CodexTurnMetadataEventRequest { event_params: CodexTurnMetadata, } +#[derive(Serialize)] +struct CodexInputMessageMetadata { + thread_id: Option, + turn_id: Option, + product_client_id: Option, + model_slug: Option, + message_role: Option, + user_message_type: Option, +} + +#[derive(Serialize)] +struct CodexInputMessageMetadataEventRequest { + event_type: &'static str, + event_params: CodexInputMessageMetadata, +} + #[derive(Serialize)] struct CodexPluginMetadata { plugin_id: Option, @@ -523,6 +576,26 @@ pub(crate) fn track_turn_metadata( queue.try_send(job); } +pub(crate) fn track_input_message_metadata( + queue: &AnalyticsEventsQueue, + config: Arc, + tracking: Option, + input_message_metadata: InputMessageMetadata, +) { + if config.analytics_enabled == Some(false) { + return; + } + let Some(tracking) = tracking else { + return; + }; + let job = TrackEventsJob::InputMessageMetadata(TrackInputMessageMetadataJob { + config, + tracking, + input_message_metadata, + }); + queue.try_send(job); +} + pub(crate) fn track_plugin_used( queue: &AnalyticsEventsQueue, config: Arc, @@ -664,6 +737,25 @@ async fn send_track_turn_metadata(auth_manager: &AuthManager, job: TrackTurnMeta send_track_events(auth_manager, config, events).await; } +async fn send_track_input_message_metadata( + auth_manager: &AuthManager, + job: TrackInputMessageMetadataJob, +) { + let TrackInputMessageMetadataJob { + config, + tracking, + input_message_metadata, + } = job; + let events = vec![TrackEventRequest::InputMessageMetadata( + CodexInputMessageMetadataEventRequest { + event_type: "codex_input_message_metadata", + event_params: codex_input_message_metadata(&tracking, input_message_metadata), + }, + )]; + + send_track_events(auth_manager, config, events).await; +} + async fn send_track_plugin_used(auth_manager: &AuthManager, job: TrackPluginUsedJob) { let TrackPluginUsedJob { config, @@ -747,6 +839,20 @@ fn codex_turn_metadata( } } +fn codex_input_message_metadata( + tracking: &TrackEventsContext, + input_message_metadata: InputMessageMetadata, +) -> 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), + } +} + fn sandbox_policy_mode(sandbox_policy: &SandboxPolicy) -> &'static str { match sandbox_policy { SandboxPolicy::DangerFullAccess => "full_access", diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index bff12cc78f..b7dbf70181 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -2,14 +2,19 @@ use super::AnalyticsEventsQueue; use super::AppInvocation; use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; +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::UserMessageType; use super::codex_app_metadata; +use super::codex_input_message_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; use super::codex_turn_metadata; @@ -235,6 +240,42 @@ fn turn_metadata_event_serializes_expected_shape() { ); } +#[test] +fn input_message_metadata_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::InputMessageMetadata(CodexInputMessageMetadataEventRequest { + event_type: "codex_input_message_metadata", + event_params: codex_input_message_metadata( + &tracking, + InputMessageMetadata { + message_role: InputMessageRole::User, + user_message_type: UserMessageType::PromptSteering, + }, + ), + }); + + let payload = serde_json::to_value(&event).expect("serialize input message metadata event"); + + assert_eq!( + payload, + json!({ + "event_type": "codex_input_message_metadata", + "event_params": { + "thread_id": "thread-2", + "turn_id": "turn-2", + "product_client_id": crate::default_client::originator().value, + "model_slug": "gpt-5", + "message_role": "user", + "user_message_type": "prompt_steering" + } + }) + ); +} + #[test] fn plugin_used_event_serializes_expected_shape() { let tracking = TrackEventsContext { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 1100b7b640..4051a455a5 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -14,8 +14,11 @@ 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::build_track_events_context; use crate::apps::render_apps_section; use crate::auth_env_telemetry::collect_auth_env_telemetry; @@ -107,6 +110,7 @@ use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::TurnContextNetworkItem; +use codex_protocol::protocol::UserMessageType as SubmittedUserMessageType; use codex_protocol::protocol::W3cTraceContext; use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; @@ -848,6 +852,7 @@ pub(crate) struct TurnContext { pub(crate) user_instructions: Option, pub(crate) collaboration_mode: CollaborationMode, pub(crate) personality: Option, + pub(crate) user_message_type: Option, pub(crate) approval_policy: Constrained, pub(crate) sandbox_policy: Constrained, pub(crate) file_system_sandbox_policy: FileSystemSandboxPolicy, @@ -1185,6 +1190,7 @@ pub(crate) struct SessionSettingsUpdate { pub(crate) final_output_json_schema: Option>, pub(crate) personality: Option, pub(crate) app_server_client_name: Option, + pub(crate) user_message_type: Option, } impl Session { @@ -1403,6 +1409,7 @@ impl Session { user_instructions: session_configuration.user_instructions.clone(), collaboration_mode: session_configuration.collaboration_mode.clone(), personality: session_configuration.personality, + user_message_type: None, approval_policy: session_configuration.approval_policy.clone(), sandbox_policy: session_configuration.sandbox_policy.clone(), file_system_sandbox_policy: session_configuration.file_system_sandbox_policy.clone(), @@ -2394,6 +2401,7 @@ impl Session { sub_id, session_configuration, updates.final_output_json_schema, + updates.user_message_type, sandbox_policy_changed, ) .await) @@ -2404,6 +2412,7 @@ impl Session { sub_id: String, session_configuration: SessionConfiguration, final_output_json_schema: Option>, + user_message_type: Option, sandbox_policy_changed: bool, ) -> Arc { let per_turn_config = Self::build_per_turn_config(&session_configuration); @@ -2471,6 +2480,7 @@ impl Session { if let Some(final_schema) = final_output_json_schema { turn_context.final_output_json_schema = final_schema; } + turn_context.user_message_type = user_message_type; let turn_context = Arc::new(turn_context); turn_context.turn_metadata_state.spawn_git_enrichment_task(); turn_context @@ -2575,6 +2585,7 @@ impl Session { sub_id, session_configuration, /*final_output_json_schema*/ None, + /*user_message_type*/ None, /*sandbox_policy_changed*/ false, ) .await @@ -2739,6 +2750,16 @@ impl Session { )) } + async fn active_turn_tracking(&self) -> Option { + let active = self.active_turn.lock().await; + let (_, task) = active.as_ref()?.tasks.first()?; + Some(build_track_events_context( + task.turn_context.model_info.slug.clone(), + self.conversation_id.to_string(), + task.turn_context.sub_id.clone(), + )) + } + pub(crate) async fn record_execpolicy_amendment_message( &self, sub_id: &str, @@ -3886,6 +3907,7 @@ impl Session { let Some((active_turn_id, _)) = active_turn.tasks.first() else { return Err(SteerInputError::NoActiveTurn(input)); }; + let active_turn_id = active_turn_id.clone(); if let Some(expected_turn_id) = expected_turn_id && expected_turn_id != active_turn_id @@ -3913,7 +3935,20 @@ impl Session { let mut turn_state = active_turn.turn_state.lock().await; turn_state.push_pending_input(input.into()); - Ok(active_turn_id.clone()) + drop(turn_state); + drop(active); + if let Some(tracking) = self.active_turn_tracking().await { + self.services + .analytics_events_client + .track_input_message_metadata( + tracking, + InputMessageMetadata { + message_role: InputMessageRole::User, + user_message_type: UserMessageType::PromptSteering, + }, + ); + } + Ok(active_turn_id) } /// Returns the input if there was no task running to inject into. @@ -4303,7 +4338,7 @@ async fn submission_loop(sess: Arc, config: Arc, rx_sub: Receiv .await; false } - Op::UserInput { .. } | Op::UserTurn { .. } => { + Op::UserInput { .. } | Op::UserInputWithMetadata { .. } | Op::UserTurn { .. } => { handlers::user_input_or_turn(&sess, sub.id.clone(), sub.op).await; false } @@ -4586,6 +4621,7 @@ mod handlers { final_output_json_schema: Some(final_output_json_schema), personality, app_server_client_name: None, + user_message_type: None, }, ) } @@ -4596,6 +4632,19 @@ mod handlers { items, SessionSettingsUpdate { final_output_json_schema: Some(final_output_json_schema), + user_message_type: None, + ..Default::default() + }, + ), + Op::UserInputWithMetadata { + items, + final_output_json_schema, + user_message_type, + } => ( + items, + SessionSettingsUpdate { + final_output_json_schema: Some(final_output_json_schema), + user_message_type, ..Default::default() }, ), @@ -5662,6 +5711,18 @@ pub(crate) async fn run_turn( user_prompt_submit_outcome.additional_contexts }; if !input.is_empty() { + 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 { diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index dd214995e0..344dc2a1b6 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -33,6 +33,7 @@ use core_test_support::responses::ev_response_created; use core_test_support::responses::ev_web_search_call_added_partial; use core_test_support::responses::ev_web_search_call_done; use core_test_support::responses::mount_sse_once; +use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; @@ -46,6 +47,7 @@ use std::path::Path; use std::path::PathBuf; use std::time::Duration; use std::time::Instant; +use tempfile::tempdir; fn image_generation_artifact_path(codex_home: &Path, session_id: &str, call_id: &str) -> PathBuf { fn sanitize(value: &str) -> String { @@ -71,6 +73,37 @@ fn image_generation_artifact_path(codex_home: &Path, session_id: &str, call_id: .join(format!("{}.png", sanitize(call_id))) } +async fn wait_for_analytics_event( + server: &wiremock::MockServer, + event_type: &str, + event_match: impl Fn(&Value) -> bool, +) -> Value { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let requests = server.received_requests().await.unwrap_or_default(); + if let Some(event) = requests + .into_iter() + .filter(|request| request.url.path() == "/codex/analytics-events/events") + .find_map(|request| { + let payload: Value = + serde_json::from_slice(&request.body).expect("analytics payload"); + payload["events"].as_array().and_then(|events| { + events + .iter() + .find(|event| event["event_type"] == event_type && event_match(event)) + .cloned() + }) + }) + { + return event; + } + if Instant::now() >= deadline { + panic!("timed out waiting for analytics event {event_type}"); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn user_message_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); @@ -225,6 +258,193 @@ async fn user_turn_tracks_turn_metadata_analytics() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn user_turn_tracks_input_message_prompt_metadata_analytics() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + mount_sse_once( + &server, + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + ) + .await; + + let chatgpt_base_url = server.uri(); + let mut builder = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(move |config| { + config.chatgpt_base_url = chatgpt_base_url; + }); + let TestCodex { + codex, + session_configured, + config, + .. + } = builder.build(&server).await?; + + codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "hello input message analytics".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: config.cwd.clone(), + approval_policy: AskForApproval::Never, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + model: session_configured.model.clone(), + effort: Some(ReasoningEffort::High), + summary: Some(ReasoningSummary::Detailed), + service_tier: Some(Some(ServiceTier::Flex)), + collaboration_mode: Some(CollaborationMode { + mode: ModeKind::Plan, + settings: Settings { + model: session_configured.model.clone(), + reasoning_effort: Some(ReasoningEffort::High), + developer_instructions: None, + }, + }), + personality: None, + }) + .await?; + + 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" + }) + .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["product_client_id"], + serde_json::json!(codex_core::default_client::originator().value) + ); + assert_eq!(event_params["model_slug"], session_configured.model); + assert!(event_params["thread_id"].as_str().is_some()); + assert!(event_params["turn_id"].as_str().is_some()); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn user_turn_tracks_input_message_prompt_steering_metadata_analytics() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let temp = tempdir()?; + let unblock_path = temp.path().join("unblock-steering"); + let command = format!( + "while [ ! -f \"{}\" ]; do sleep 0.01; done; echo done", + unblock_path.display() + ); + let call_id = "shell-steering-call"; + + mount_sse_sequence( + &server, + vec![ + sse(vec![ + ev_response_created("resp-1"), + core_test_support::responses::ev_function_call( + call_id, + "shell", + &serde_json::to_string(&serde_json::json!({ + "command": ["/bin/sh", "-c", command], + }))?, + ), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_assistant_message("msg-2", "done"), + ev_completed("resp-2"), + ]), + ], + ) + .await; + + let chatgpt_base_url = server.uri(); + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model("gpt-5") + .with_config(move |config| { + config.chatgpt_base_url = chatgpt_base_url; + }) + .build(&server) + .await?; + let codex = test.codex.clone(); + let turn_model = test.session_configured.model.clone(); + + codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "start steering flow".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + cwd: test.cwd_path().to_path_buf(), + approval_policy: AskForApproval::Never, + approvals_reviewer: None, + sandbox_policy: SandboxPolicy::DangerFullAccess, + model: turn_model, + effort: None, + summary: None, + service_tier: None, + collaboration_mode: None, + personality: None, + }) + .await?; + + let turn_id = wait_for_event_match(&codex, |ev| match ev { + EventMsg::TurnStarted(event) => Some(event.turn_id.clone()), + _ => None, + }) + .await; + + wait_for_event_match(&codex, |ev| match ev { + EventMsg::ExecCommandBegin(event) if event.call_id == call_id => Some(()), + _ => None, + }) + .await; + + let steering_text = "steering metadata check"; + let steered_turn_id = codex + .steer_input( + vec![UserInput::Text { + text: steering_text.into(), + text_elements: Vec::new(), + }], + Some(turn_id.as_str()), + ) + .await + .expect("steer should succeed on active turn"); + assert_eq!(steered_turn_id, turn_id); + + std::fs::write(&unblock_path, "go")?; + + 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_steering" + }) + .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"], + serde_json::json!(codex_core::default_client::originator().value) + ); + assert_eq!(event_params["model_slug"], "gpt-5"); + assert!(event_params["thread_id"].as_str().is_some()); + assert!(event_params["turn_id"].as_str().is_some()); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn assistant_message_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); diff --git a/codex-rs/debug-client/src/client.rs b/codex-rs/debug-client/src/client.rs index 762c11cb63..073ec407f3 100644 --- a/codex-rs/debug-client/src/client.rs +++ b/codex-rs/debug-client/src/client.rs @@ -199,6 +199,7 @@ impl AppServerClient { // Debug client sends plain text with no UI markup spans. text_elements: Vec::new(), }], + user_message_type: None, ..Default::default() }, }; diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index c10b86f696..e3c094b85a 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -708,6 +708,7 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { summary: None, personality: None, output_schema, + user_message_type: None, collaboration_mode: None, }, }, diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index b5c96f4d05..e92e62ca53 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -204,6 +204,14 @@ pub struct ConversationTextParams { pub text: String, } +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "snake_case")] +pub enum UserMessageType { + Prompt, + PromptSteering, + PromptQueued, +} + /// Submission operation #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema)] #[serde(tag = "type", rename_all = "snake_case")] @@ -242,6 +250,19 @@ pub enum Op { final_output_json_schema: Option, }, + /// User input with additional turn-scoped metadata supplied by a higher-level + /// surface such as app-server `turn/start`. + UserInputWithMetadata { + /// User input items, see `InputItem` + items: Vec, + /// Optional JSON Schema used to constrain the final assistant message for this turn. + #[serde(skip_serializing_if = "Option::is_none")] + final_output_json_schema: Option, + /// Semantic classification of this submitted user input. + #[serde(default, skip_serializing_if = "Option::is_none")] + user_message_type: Option, + }, + /// Similar to [`Op::UserInput`], but contains additional context required /// for a turn of a [`crate::codex_thread::CodexThread`]. UserTurn { @@ -568,6 +589,7 @@ impl Op { Self::RealtimeConversationText(_) => "realtime_conversation_text", Self::RealtimeConversationClose => "realtime_conversation_close", Self::UserInput { .. } => "user_input", + Self::UserInputWithMetadata { .. } => "user_input", Self::UserTurn { .. } => "user_turn", Self::InterAgentCommunication { .. } => "inter_agent_communication", Self::OverrideTurnContext { .. } => "override_turn_context", 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 0da85bc617..537325d0f5 100644 --- a/codex-rs/tui_app_server/src/app_server_session.rs +++ b/codex-rs/tui_app_server/src/app_server_session.rs @@ -397,6 +397,7 @@ impl AppServerSession { effort, summary, personality, + user_message_type: None, output_schema, collaboration_mode, },