From 07c0d34fa43b0f7ba3850006a0260f664d67903c 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 | 190 +++++++++++---- codex-rs/core/src/analytics_client_tests.rs | 61 ++++- codex-rs/core/src/codex.rs | 175 ++++++++------ 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, 659 insertions(+), 118 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 09647e05cc..51f0f7acb0 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; @@ -3925,6 +3926,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)] @@ -8044,9 +8049,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 32eb28c12b..72fdb676cf 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -6112,9 +6112,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 aa66f79f29..98313e6521 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 a81c87bd4e..ec251b40d1 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 2b5d1fe8a0..fff7397895 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -217,6 +217,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?; @@ -239,6 +240,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 e5b5f13421..b15179acd3 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -13,6 +13,7 @@ use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SkillScope; +use codex_protocol::protocol::UserMessageType; use serde::Serialize; use sha1::Digest; use sha1::Sha1; @@ -32,7 +33,7 @@ pub(crate) struct TrackEventsContext { } #[derive(Clone)] -pub(crate) struct CodexTurnEvent { +pub(crate) struct TurnMetadata { pub(crate) model_provider: String, pub(crate) sandbox_policy: SandboxPolicy, pub(crate) reasoning_effort: Option, @@ -46,6 +47,18 @@ pub(crate) struct CodexTurnEvent { 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) fn build_track_events_context( model_slug: String, thread_id: String, @@ -107,8 +120,11 @@ impl AnalyticsEventsQueue { TrackEventsJob::AppUsed(job) => { send_track_app_used(&auth_manager, job).await; } - TrackEventsJob::TurnEvent(job) => { - send_track_turn_event(&auth_manager, job).await; + 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; @@ -223,16 +239,29 @@ impl AnalyticsEventsClient { ); } - pub(crate) fn track_turn_event( + pub(crate) fn track_turn_metadata( &self, tracking: TrackEventsContext, - turn_event: CodexTurnEvent, + turn_metadata: TurnMetadata, ) { - track_turn_event( + track_turn_metadata( &self.queue, Arc::clone(&self.config), Some(tracking), - turn_event, + turn_metadata, + ); + } + + 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, ); } @@ -277,7 +306,8 @@ enum TrackEventsJob { SkillInvocations(TrackSkillInvocationsJob), AppMentioned(TrackAppMentionedJob), AppUsed(TrackAppUsedJob), - TurnEvent(TrackTurnEventJob), + TurnMetadata(TrackTurnMetadataJob), + InputMessageMetadata(TrackInputMessageMetadataJob), PluginUsed(TrackPluginUsedJob), PluginInstalled(TrackPluginManagementJob), PluginUninstalled(TrackPluginManagementJob), @@ -303,10 +333,16 @@ struct TrackAppUsedJob { app: AppInvocation, } -struct TrackTurnEventJob { +struct TrackTurnMetadataJob { config: Arc, tracking: TrackEventsContext, - turn_event: CodexTurnEvent, + turn_metadata: TurnMetadata, +} + +struct TrackInputMessageMetadataJob { + config: Arc, + tracking: TrackEventsContext, + input_message_metadata: InputMessageMetadata, } struct TrackPluginUsedJob { @@ -343,7 +379,8 @@ enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), - TurnEvent(CodexTurnEventRequest), + TurnMetadata(CodexTurnMetadataEventRequest), + InputMessageMetadata(CodexInputMessageMetadataEventRequest), PluginUsed(CodexPluginUsedEventRequest), PluginInstalled(CodexPluginEventRequest), PluginUninstalled(CodexPluginEventRequest), @@ -393,7 +430,7 @@ struct CodexAppUsedEventRequest { } #[derive(Serialize)] -struct CodexTurnEventParams { +struct CodexTurnMetadata { thread_id: String, turn_id: String, product_client_id: Option, @@ -412,9 +449,25 @@ struct CodexTurnEventParams { } #[derive(Serialize)] -struct CodexTurnEventRequest { +struct CodexTurnMetadataEventRequest { event_type: &'static str, - event_params: CodexTurnEventParams, + 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)] @@ -518,11 +571,11 @@ pub(crate) fn track_app_used( queue.try_send(job); } -pub(crate) fn track_turn_event( +pub(crate) fn track_turn_metadata( queue: &AnalyticsEventsQueue, config: Arc, tracking: Option, - turn_event: CodexTurnEvent, + turn_metadata: TurnMetadata, ) { if config.analytics_enabled == Some(false) { return; @@ -530,10 +583,30 @@ pub(crate) fn track_turn_event( let Some(tracking) = tracking else { return; }; - let job = TrackEventsJob::TurnEvent(TrackTurnEventJob { + let job = TrackEventsJob::TurnMetadata(TrackTurnMetadataJob { config, tracking, - turn_event, + 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); } @@ -663,16 +736,37 @@ async fn send_track_app_used(auth_manager: &AuthManager, job: TrackAppUsedJob) { send_track_events(auth_manager, config, events).await; } -async fn send_track_turn_event(auth_manager: &AuthManager, job: TrackTurnEventJob) { - let TrackTurnEventJob { +async fn send_track_turn_metadata(auth_manager: &AuthManager, job: TrackTurnMetadataJob) { + let TrackTurnMetadataJob { config, tracking, - turn_event, + turn_metadata, } = job; - let events = vec![TrackEventRequest::TurnEvent(CodexTurnEventRequest { - event_type: "codex_turn_event", - event_params: codex_turn_event_params(&tracking, turn_event), - })]; + let events = vec![TrackEventRequest::TurnMetadata( + CodexTurnMetadataEventRequest { + event_type: "codex_turn_metadata", + event_params: codex_turn_metadata(&tracking, turn_metadata), + }, + )]; + + 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; } @@ -741,29 +835,45 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } } -fn codex_turn_event_params( +fn codex_turn_metadata( tracking: &TrackEventsContext, - turn_event: CodexTurnEvent, -) -> CodexTurnEventParams { - CodexTurnEventParams { + turn_metadata: TurnMetadata, +) -> CodexTurnMetadata { + CodexTurnMetadata { thread_id: tracking.thread_id.clone(), turn_id: tracking.turn_id.clone(), product_client_id: Some(crate::default_client::originator().value), model: Some(tracking.model_slug.clone()), - model_provider: turn_event.model_provider, - sandbox_policy: Some(sandbox_policy_mode(&turn_event.sandbox_policy)), - reasoning_effort: turn_event.reasoning_effort.map(|value| value.to_string()), - reasoning_summary: reasoning_summary_mode(turn_event.reasoning_summary), - service_tier: turn_event + model_provider: turn_metadata.model_provider, + sandbox_policy: Some(sandbox_policy_mode(&turn_metadata.sandbox_policy)), + reasoning_effort: turn_metadata + .reasoning_effort + .map(|value| value.to_string()), + reasoning_summary: reasoning_summary_mode(turn_metadata.reasoning_summary), + service_tier: turn_metadata .service_tier .map(|value| value.to_string()) .unwrap_or_else(|| "default".to_string()), - approval_policy: turn_event.approval_policy.to_string(), - approvals_reviewer: turn_event.approvals_reviewer.to_string(), - sandbox_network_access: turn_event.sandbox_network_access, - collaboration_mode: Some(collaboration_mode_mode(turn_event.collaboration_mode)), - personality: personality_mode(turn_event.personality), - num_input_images: turn_event.num_input_images, + approval_policy: turn_metadata.approval_policy.to_string(), + approvals_reviewer: turn_metadata.approvals_reviewer.to_string(), + sandbox_network_access: turn_metadata.sandbox_network_access, + collaboration_mode: Some(collaboration_mode_mode(turn_metadata.collaboration_mode)), + personality: personality_mode(turn_metadata.personality), + num_input_images: turn_metadata.num_input_images, + } +} + +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), } } diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index 1a8cb80051..5329cac77d 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -2,17 +2,22 @@ use super::AnalyticsEventsQueue; use super::AppInvocation; use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; +use super::CodexInputMessageMetadataEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; -use super::CodexTurnEvent; -use super::CodexTurnEventRequest; +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_event_params; +use super::codex_turn_metadata; use super::normalize_path_for_skill_id; use crate::plugins::AppConnectorId; use crate::plugins::PluginCapabilitySummary; @@ -197,17 +202,17 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { } #[test] -fn turn_event_serializes_expected_shape() { +fn turn_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::TurnEvent(CodexTurnEventRequest { - event_type: "codex_turn_event", - event_params: codex_turn_event_params( + let event = TrackEventRequest::TurnMetadata(CodexTurnMetadataEventRequest { + event_type: "codex_turn_metadata", + event_params: codex_turn_metadata( &tracking, - CodexTurnEvent { + TurnMetadata { model_provider: "openai".to_string(), sandbox_policy: SandboxPolicy::new_read_only_policy(), reasoning_effort: Some(ReasoningEffort::High), @@ -223,12 +228,12 @@ fn turn_event_serializes_expected_shape() { ), }); - let payload = serde_json::to_value(&event).expect("serialize turn event"); + let payload = serde_json::to_value(&event).expect("serialize turn metadata event"); assert_eq!( payload, json!({ - "event_type": "codex_turn_event", + "event_type": "codex_turn_metadata", "event_params": { "thread_id": "thread-2", "turn_id": "turn-2", @@ -250,6 +255,42 @@ fn turn_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 5327771caa..e60ea1d581 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::CodexTurnEvent; +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; @@ -236,6 +240,9 @@ pub(crate) struct PreviousTurnSettings { use crate::exec_policy::ExecPolicyUpdateError; use crate::feedback_tags; +use crate::file_watcher::FileWatcher; +use crate::file_watcher::FileWatcherEvent; +use crate::git_info::get_git_repo_root; use crate::guardian::GuardianReviewSessionManager; use crate::hook_runtime::PendingInputHookDisposition; use crate::hook_runtime::inspect_pending_input; @@ -320,8 +327,6 @@ use crate::skills::injection::ToolMentionKind; use crate::skills::injection::app_id_from_path; use crate::skills::injection::tool_kind_for_path; use crate::skills::resolve_skill_dependencies_for_turn; -use crate::skills_watcher::SkillsWatcher; -use crate::skills_watcher::SkillsWatcherEvent; use crate::state::ActiveTurn; use crate::state::SessionServices; use crate::state::SessionState; @@ -350,7 +355,6 @@ use crate::unified_exec::UnifiedExecProcessManager; use crate::util::backoff; use crate::windows_sandbox::WindowsSandboxLevelExt; use codex_async_utils::OrCancelExt; -use codex_git_utils::get_git_repo_root; use codex_otel::SessionTelemetry; use codex_otel::TelemetryAuthMode; use codex_otel::metrics::names::THREAD_STARTED_METRIC; @@ -404,7 +408,7 @@ pub(crate) struct CodexSpawnArgs { pub(crate) skills_manager: Arc, pub(crate) plugins_manager: Arc, pub(crate) mcp_manager: Arc, - pub(crate) skills_watcher: Arc, + pub(crate) file_watcher: Arc, pub(crate) conversation_history: InitialHistory, pub(crate) session_source: SessionSource, pub(crate) agent_control: AgentControl, @@ -457,7 +461,7 @@ impl Codex { skills_manager, plugins_manager, mcp_manager, - skills_watcher, + file_watcher, conversation_history, session_source, agent_control, @@ -645,7 +649,7 @@ impl Codex { skills_manager, plugins_manager, mcp_manager.clone(), - skills_watcher, + file_watcher, agent_control, ) .await @@ -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 { @@ -1298,13 +1304,13 @@ impl Session { self.out_of_band_elicitation_paused.send_replace(paused); } - fn start_skills_watcher_listener(self: &Arc) { - let mut rx = self.services.skills_watcher.subscribe(); + fn start_file_watcher_listener(self: &Arc) { + let mut rx = self.services.file_watcher.subscribe(); let weak_sess = Arc::downgrade(self); tokio::spawn(async move { loop { match rx.recv().await { - Ok(SkillsWatcherEvent::SkillsChanged { .. }) => { + Ok(FileWatcherEvent::SkillsChanged { .. }) => { let Some(sess) = weak_sess.upgrade() else { break; }; @@ -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(), @@ -1440,7 +1447,7 @@ impl Session { skills_manager: Arc, plugins_manager: Arc, mcp_manager: Arc, - skills_watcher: Arc, + file_watcher: Arc, agent_control: AgentControl, ) -> anyhow::Result> { debug!( @@ -1859,7 +1866,7 @@ impl Session { skills_manager, plugins_manager: Arc::clone(&plugins_manager), mcp_manager: Arc::clone(&mcp_manager), - skills_watcher, + file_watcher, agent_control, network_proxy, network_approval: Arc::clone(&network_approval), @@ -1938,7 +1945,7 @@ impl Session { } // Start the watcher after SessionConfigured so it cannot emit earlier events. - sess.start_skills_watcher_listener(); + sess.start_file_watcher_listener(); // Construct sandbox_state before MCP startup so it can be sent to each // MCP server immediately after it becomes ready (avoiding blocking). let sandbox_state = SandboxState { @@ -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. @@ -3957,17 +3992,6 @@ impl Session { } } - pub(crate) async fn pending_input_snapshot(&self) -> Vec { - let active = self.active_turn.lock().await; - match active.as_ref() { - Some(at) => { - let ts = at.turn_state.lock().await; - ts.pending_input_snapshot() - } - None => Vec::with_capacity(0), - } - } - /// Queue response items to be injected into the next active turn created for this session. pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec) { if items.is_empty() { @@ -3982,12 +4006,6 @@ impl Session { std::mem::take(&mut *self.idle_pending_input.lock().await) } - pub(crate) async fn queued_response_items_for_next_turn_snapshot( - &self, - ) -> Vec { - self.idle_pending_input.lock().await.clone() - } - pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool { !self.idle_pending_input.lock().await.is_empty() } @@ -4320,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 } @@ -4603,6 +4621,7 @@ mod handlers { final_output_json_schema: Some(final_output_json_schema), personality, app_server_client_name: None, + user_message_type: None, }, ) } @@ -4613,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() }, ), @@ -4651,27 +4683,27 @@ mod handlers { } } - /// Records an inter-agent assistant envelope and, when requested, wakes the recipient by - /// starting a regular turn if the session is currently idle. pub async fn inter_agent_communication( sess: &Arc, sub_id: String, communication: InterAgentCommunication, ) { let pending_item = communication.to_response_input_item(); - if let Ok(()) = sess.inject_response_items(vec![pending_item.clone()]).await { + if sess + .inject_response_items(vec![pending_item.clone()]) + .await + .is_ok() + { return; } + let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; + sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) + .await; sess.queue_response_items_for_next_turn(vec![pending_item]) .await; - if communication.trigger_turn { - let turn_context = sess.new_default_turn_with_sub_id(sub_id).await; - sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref()) - .await; - sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new()) - .await; - } + sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new()) + .await; } pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command: String) { @@ -5678,6 +5710,41 @@ pub(crate) async fn run_turn( .await; 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 { + model_provider: turn_context.config.model_provider_id.clone(), + sandbox_policy: turn_context.sandbox_policy.get().clone(), + reasoning_effort: turn_context.reasoning_effort, + reasoning_summary: Some(turn_context.reasoning_summary), + service_tier: turn_context.config.service_tier, + approval_policy: turn_context.approval_policy.value(), + approvals_reviewer: turn_context.config.approvals_reviewer, + sandbox_network_access: turn_context.network_sandbox_policy.is_enabled(), + collaboration_mode: turn_context.collaboration_mode.mode, + personality: turn_context.personality, + num_input_images: input + .iter() + .filter(|item| { + matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. }) + }) + .count(), + }, + ); + } sess.services .analytics_events_client .track_app_mentioned(tracking.clone(), mentioned_app_invocations); @@ -5991,30 +6058,6 @@ pub(crate) async fn run_turn( } } - if !input.is_empty() { - sess.services.analytics_events_client.track_turn_event( - tracking, - CodexTurnEvent { - model_provider: turn_context.config.model_provider_id.clone(), - sandbox_policy: turn_context.sandbox_policy.get().clone(), - reasoning_effort: turn_context.reasoning_effort, - reasoning_summary: Some(turn_context.reasoning_summary), - service_tier: turn_context.config.service_tier, - approval_policy: turn_context.approval_policy.value(), - approvals_reviewer: turn_context.config.approvals_reviewer, - sandbox_network_access: turn_context.network_sandbox_policy.is_enabled(), - collaboration_mode: turn_context.collaboration_mode.mode, - personality: turn_context.personality, - num_input_images: input - .iter() - .filter(|item| { - matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. }) - }) - .count(), - }, - ); - } - last_agent_message } diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index f34fcca1de..855b14dcf0 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(())); @@ -237,6 +270,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 c4f106b464..86e707ab07 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -674,6 +674,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 c40514e3db..284b78674b 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -205,6 +205,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")] @@ -243,6 +251,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 { @@ -572,6 +593,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 ecdea588ee..8274bda7a2 100644 --- a/codex-rs/tui_app_server/src/app_server_session.rs +++ b/codex-rs/tui_app_server/src/app_server_session.rs @@ -407,6 +407,7 @@ impl AppServerSession { effort, summary, personality, + user_message_type: None, output_schema, collaboration_mode, },