rework to isolate core

codex-rs/core/src/codex.rs

@@ -83,7 +83,6 @@ use codex_protocol::dynamic_tools::DynamicToolResponse;
 use codex_protocol::dynamic_tools::DynamicToolSpec;
 use codex_protocol::items::PlanItem;
 use codex_protocol::items::TurnItem;
-use codex_protocol::items::TurnItemMetadata;
 use codex_protocol::items::UserMessageItem;
 use codex_protocol::items::UserMessageType;
 use codex_protocol::mcp::CallToolResult;
@@ -995,6 +994,18 @@ fn local_time_context() -> (String, String) {
     }
 }
 
+fn stamp_user_message_type_on_input_item(item: &mut ResponseInputItem, kind: UserMessageType) {
+    let ResponseInputItem::Message { role, metadata, .. } = item else {
+        return;
+    };
+    if role != "user" {
+        return;
+    }
+    let mut metadata_value = metadata.take().unwrap_or_default();
+    metadata_value.user_message_type = Some(kind);
+    *metadata = Some(metadata_value);
+}
+
 #[derive(Clone)]
 pub(crate) struct SessionConfiguration {
     /// Provider identifier ("openai", "openrouter", ...).
@@ -2686,6 +2697,7 @@ impl Session {
             .inject_response_items(vec![ResponseInputItem::Message {
                 role: "developer".to_string(),
                 content: vec![ContentItem::InputText { text }],
+                metadata: None,
             }])
             .await
             .is_err()
@@ -2783,6 +2795,7 @@ impl Session {
             .inject_response_items(vec![ResponseInputItem::Message {
                 role: "developer".to_string(),
                 content: vec![ContentItem::InputText { text }],
+                metadata: None,
             }])
             .await
             .is_err()
@@ -3714,10 +3727,7 @@ impl Session {
         // those spans, and `record_response_item_and_emit_turn_item` would drop them.
         self.record_conversation_items(turn_context, std::slice::from_ref(&response_item))
             .await;
-        let metadata = user_message_type.map(|kind| TurnItemMetadata {
-            user_message_type: Some(kind),
-        });
-        let turn_item = TurnItem::UserMessage(UserMessageItem::new_with_metadata(input, metadata));
+        let turn_item = TurnItem::UserMessage(UserMessageItem::new(input));
         self.emit_turn_item_started(turn_context, &turn_item).await;
         self.emit_turn_item_completed(turn_context, turn_item).await;
         self.ensure_rollout_materialized().await;
@@ -3810,8 +3820,13 @@ impl Session {
             });
         }
 
+        let mut input_item: ResponseInputItem = input.into();
+        if self.enabled(Feature::UserMessageTypeMetadata) {
+            stamp_user_message_type_on_input_item(&mut input_item, UserMessageType::PromptSteering);
+        }
+
         let mut turn_state = active_turn.turn_state.lock().await;
-        turn_state.push_pending_input(input.into(), Some(UserMessageType::PromptSteering));
+        turn_state.push_pending_input(input_item, Some(UserMessageType::PromptSteering));
         Ok(active_turn_id.clone())
     }
 
@@ -3824,11 +3839,16 @@ impl Session {
         match active.as_mut() {
             Some(at) => {
                 let mut ts = at.turn_state.lock().await;
-                for item in input {
+                for mut item in input {
                     let user_message_type = match &item {
                         ResponseInputItem::Message { .. } => Some(UserMessageType::PromptQueued),
                         _ => None,
                     };
+                    if self.enabled(Feature::UserMessageTypeMetadata)
+                        && let Some(kind) = user_message_type.clone()
+                    {
+                        stamp_user_message_type_on_input_item(&mut item, kind);
+                    }
                     ts.push_pending_input(item, user_message_type);
                 }
                 Ok(())
@@ -5577,7 +5597,10 @@ pub(crate) async fn run_turn(
     sess.merge_connector_selection(explicitly_enabled_connectors.clone())
         .await;
 
-    let initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input.clone());
+    let mut initial_input_for_turn: ResponseInputItem = ResponseInputItem::from(input.clone());
+    if sess.enabled(Feature::UserMessageTypeMetadata) {
+        stamp_user_message_type_on_input_item(&mut initial_input_for_turn, UserMessageType::Prompt);
+    }
     let response_item: ResponseItem = initial_input_for_turn.clone().into();
     sess.record_user_prompt_and_emit_turn_item(
         turn_context.as_ref(),

codex-rs/core/src/codex_tests.rs

@@ -3707,6 +3707,7 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input()
         content: vec![ContentItem::InputText {
             text: "late pending input".to_string(),
         }],
+        metadata: None,
     }])
     .await
     .expect("inject pending input into active turn");
@@ -3825,6 +3826,7 @@ async fn task_finish_emits_prompt_queued_metadata_for_injected_user_input_when_f
         content: vec![ContentItem::InputText {
             text: "late queued pending input".to_string(),
         }],
+        metadata: None,
     }])
     .await
     .expect("inject pending input into active turn");
@@ -3835,7 +3837,18 @@ async fn task_finish_emits_prompt_queued_metadata_for_injected_user_input_when_f
         .await
         .expect("expected raw response item event")
         .expect("channel open");
-    assert!(matches!(first.msg, EventMsg::RawResponseItem(_)));
+    assert!(matches!(
+        first.msg,
+        EventMsg::RawResponseItem(ref event)
+            if matches!(
+                &event.item,
+                ResponseItem::Message {
+                    metadata: Some(metadata),
+                    ..
+                } if metadata.user_message_type
+                    == Some(codex_protocol::items::UserMessageType::PromptQueued)
+            )
+    ));
 
     let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
         .await
@@ -3844,10 +3857,12 @@ async fn task_finish_emits_prompt_queued_metadata_for_injected_user_input_when_f
     assert!(matches!(
         second.msg,
         EventMsg::ItemStarted(ItemStartedEvent {
-            item: TurnItem::UserMessage(UserMessageItem { metadata: Some(metadata), .. }),
+            item: TurnItem::UserMessage(UserMessageItem { content, .. }),
             ..
-        }) if metadata.user_message_type
-            == Some(codex_protocol::items::UserMessageType::PromptQueued)
+        }) if content == vec![UserInput::Text {
+            text: "late queued pending input".to_string(),
+            text_elements: Vec::new(),
+        }]
     ));
 
     let third = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv())
@@ -3857,10 +3872,12 @@ async fn task_finish_emits_prompt_queued_metadata_for_injected_user_input_when_f
     assert!(matches!(
         third.msg,
         EventMsg::ItemCompleted(ItemCompletedEvent {
-            item: TurnItem::UserMessage(UserMessageItem { metadata: Some(metadata), .. }),
+            item: TurnItem::UserMessage(UserMessageItem { content, .. }),
             ..
-        }) if metadata.user_message_type
-            == Some(codex_protocol::items::UserMessageType::PromptQueued)
+        }) if content == vec![UserInput::Text {
+            text: "late queued pending input".to_string(),
+            text_elements: Vec::new(),
+        }]
     ));
 }
 

codex-rs/core/src/codex_thread.rs

@@ -137,6 +137,7 @@ impl CodexThread {
         let pending_item = ResponseInputItem::Message {
             role: "user".to_string(),
             content: vec![ContentItem::InputText { text: message }],
+            metadata: None,
         };
         let pending_items = vec![pending_item];
         let Err(items_without_active_turn) = self

codex-rs/core/src/tasks/user_shell.rs

@@ -326,7 +326,16 @@ async fn persist_user_shell_output(
     }
 
     let response_input_item = match output_item {
-        ResponseItem::Message { role, content, .. } => ResponseInputItem::Message { role, content },
+        ResponseItem::Message {
+            role,
+            content,
+            metadata,
+            ..
+        } => ResponseInputItem::Message {
+            role,
+            content,
+            metadata,
+        },
         _ => unreachable!("user shell command output record should always be a message"),
     };
 

codex-rs/core/tests/suite/items.rs

@@ -175,20 +175,7 @@ async fn user_message_type_metadata_is_emitted_when_feature_enabled() -> anyhow:
     .await;
 
     assert_eq!(started_item.id, completed_item.id);
-    assert_eq!(
-        started_item
-            .metadata
-            .as_ref()
-            .and_then(|metadata| metadata.user_message_type.as_ref()),
-        Some(&codex_protocol::items::UserMessageType::Prompt)
-    );
-    assert_eq!(
-        completed_item
-            .metadata
-            .as_ref()
-            .and_then(|metadata| metadata.user_message_type.as_ref()),
-        Some(&codex_protocol::items::UserMessageType::Prompt)
-    );
+    assert_eq!(started_item.content, completed_item.content);
 
     let deadline = Instant::now() + Duration::from_secs(2);
     while req.requests().is_empty() {
@@ -449,6 +436,7 @@ async fn user_message_type_prompt_queued_metadata_is_emitted_when_feature_enable
                 content: vec![codex_protocol::models::ContentItem::InputText {
                     text: queued_text.into(),
                 }],
+                metadata: None,
             }])
             .await
             .is_ok(),

codex-rs/protocol/src/models.rs

@@ -228,6 +228,9 @@ pub enum ResponseInputItem {
     Message {
         role: String,
         content: Vec<ContentItem>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        #[ts(optional)]
+        metadata: Option<ResponseItemMetadata>,
     },
     FunctionCallOutput {
         call_id: String,
@@ -907,11 +910,15 @@ pub fn local_image_content_items_with_label_number(
 impl From<ResponseInputItem> for ResponseItem {
     fn from(item: ResponseInputItem) -> Self {
         match item {
-            ResponseInputItem::Message { role, content } => Self::Message {
+            ResponseInputItem::Message {
                 role,
                 content,
+                metadata,
+            } => Self::Message {
+                role,
+                content,
+                metadata,
                 id: None,
-                metadata: None,
                 end_turn: None,
                 phase: None,
             },
@@ -1038,6 +1045,7 @@ impl From<Vec<UserInput>> for ResponseInputItem {
                     UserInput::Skill { .. } | UserInput::Mention { .. } => Vec::new(), // Tool bodies are injected later in core
                 })
                 .collect::<Vec<ContentItem>>(),
+            metadata: None,
         }
     }
 }