From 826297a899adb5bc2eb37862e7196a5ea06cec3d Mon Sep 17 00:00:00 2001 From: Roy Han Date: Fri, 20 Mar 2026 18:03:20 -0700 Subject: [PATCH] add directly to response input item --- codex-rs/core/src/codex.rs | 56 +++++++------------- codex-rs/core/src/codex_tests.rs | 45 +++++----------- codex-rs/core/src/state/mod.rs | 1 - codex-rs/core/src/state/turn.rs | 24 ++------- codex-rs/core/src/tasks/mod.rs | 4 +- codex-rs/core/src/tools/js_repl/mod_tests.rs | 30 +++++------ codex-rs/protocol/src/item_metadata.rs | 2 +- 7 files changed, 56 insertions(+), 106 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index a2e393b6bd..4b168ea1e8 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -3314,7 +3314,9 @@ impl Session { role, content, metadata: Some( - metadata.unwrap_or_else(|| ResponseItemMessageMetadata::new(None)), + metadata.unwrap_or_else(|| { + ResponseItemMessageMetadata::new(/*user_message_type*/ None) + }), ), end_turn, phase, @@ -3913,7 +3915,7 @@ impl Session { } let mut turn_state = active_turn.turn_state.lock().await; - turn_state.push_pending_input(input_item, Some(UserMessageType::PromptSteering)); + turn_state.push_pending_input(input_item); Ok(active_turn_id.clone()) } @@ -3927,16 +3929,15 @@ impl Session { Some(at) => { let mut ts = at.turn_state.lock().await; for mut item in input { - let user_message_type = match &item { - ResponseInputItem::Message { .. } => Some(UserMessageType::PromptQueued), - _ => None, - }; if self.enabled(Feature::ItemMetadata) - && let Some(kind) = user_message_type.clone() + && matches!(&item, ResponseInputItem::Message { .. }) { - stamp_user_message_type_on_input_item(&mut item, kind); + stamp_user_message_type_on_input_item( + &mut item, + UserMessageType::PromptQueued, + ); } - ts.push_pending_input(item, user_message_type); + ts.push_pending_input(item); } Ok(()) } @@ -3944,41 +3945,26 @@ impl Session { } } - pub async fn get_pending_input_with_metadata( - &self, - ) -> Vec<(ResponseInputItem, Option)> { + pub async fn get_pending_input(&self) -> Vec { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { let mut ts = at.turn_state.lock().await; - ts.take_pending_input_with_metadata() - .into_iter() - .map(|item| (item.input, item.user_message_type)) - .collect() + ts.take_pending_input() } None => Vec::with_capacity(0), } } - pub(crate) async fn prepend_pending_input_with_metadata( + pub(crate) async fn prepend_pending_input( &self, - input: Vec<(ResponseInputItem, Option)>, + input: Vec, ) -> Result<(), ()> { let mut active = self.active_turn.lock().await; match active.as_mut() { Some(at) => { let mut ts = at.turn_state.lock().await; - ts.prepend_pending_input( - input - .into_iter() - .map( - |(input, user_message_type)| crate::state::PendingInputItem { - input, - user_message_type, - }, - ) - .collect(), - ); + ts.prepend_pending_input(input); Ok(()) } None => Err(()), @@ -5695,26 +5681,24 @@ pub(crate) async fn run_turn( // Note that pending_input would be something like a message the user // submitted through the UI while the model was running. Though the UI // may support this, the model might not. - let pending_response_items = sess.get_pending_input_with_metadata().await; + let pending_response_items = sess.get_pending_input().await; let mut blocked_pending_input = false; let mut blocked_pending_input_contexts = Vec::new(); let mut requeued_pending_input = false; let mut accepted_pending_input = Vec::new(); if !pending_response_items.is_empty() { let mut pending_input_iter = pending_response_items.into_iter(); - while let Some((pending_input_item, user_message_type)) = pending_input_iter.next() { + while let Some(pending_input_item) = pending_input_iter.next() { match inspect_pending_input(&sess, &turn_context, pending_input_item).await { PendingInputHookDisposition::Accepted(pending_input) => { - accepted_pending_input.push((*pending_input, user_message_type)); + accepted_pending_input.push(*pending_input); } PendingInputHookDisposition::Blocked { additional_contexts, } => { let remaining_pending_input = pending_input_iter.collect::>(); if !remaining_pending_input.is_empty() { - let _ = sess - .prepend_pending_input_with_metadata(remaining_pending_input) - .await; + let _ = sess.prepend_pending_input(remaining_pending_input).await; requeued_pending_input = true; } blocked_pending_input_contexts = additional_contexts; @@ -5726,7 +5710,7 @@ pub(crate) async fn run_turn( } let has_accepted_pending_input = !accepted_pending_input.is_empty(); - for (pending_input, _user_message_type) in accepted_pending_input { + for pending_input in accepted_pending_input { record_pending_input(&sess, &turn_context, pending_input).await; } record_additional_contexts(&sess, &turn_context, blocked_pending_input_contexts).await; diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 7712c4d727..f29ca4b450 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -4495,12 +4495,17 @@ async fn steer_input_returns_active_turn_id() { assert_eq!(turn_id, tc.sub_id); assert!(sess.has_pending_input().await); - let pending_input = sess.get_pending_input_with_metadata().await; + let pending_input = sess.get_pending_input().await; assert_eq!(pending_input.len(), 1); + let ResponseInputItem::Message { metadata, .. } = + pending_input.first().expect("pending input should exist") + else { + panic!("expected pending input message"); + }; assert_eq!( - pending_input - .first() - .and_then(|(_, user_message_type)| user_message_type.as_ref()), + metadata + .as_ref() + .and_then(|metadata| metadata.user_message_type.as_ref()), Some(&codex_protocol::models::UserMessageType::PromptSteering) ); } @@ -4579,20 +4584,8 @@ async fn prepend_pending_input_keeps_older_tail_ahead_of_newer_input() { .await .expect("inject initial pending input into active turn"); - let drained = sess.get_pending_input_with_metadata().await; - assert_eq!( - drained, - vec![ - ( - blocked, - Some(codex_protocol::models::UserMessageType::PromptQueued), - ), - ( - later.clone(), - Some(codex_protocol::models::UserMessageType::PromptQueued), - ), - ] - ); + let drained = sess.get_pending_input().await; + assert_eq!(drained, vec![blocked, later.clone()]); sess.inject_response_items(vec![newer.clone()]) .await @@ -4600,23 +4593,11 @@ async fn prepend_pending_input_keeps_older_tail_ahead_of_newer_input() { let mut drained_iter = drained.into_iter(); let _blocked = drained_iter.next().expect("blocked prompt should exist"); - sess.prepend_pending_input_with_metadata(drained_iter.collect()) + sess.prepend_pending_input(drained_iter.collect()) .await .expect("requeue later pending input at the front of the queue"); - assert_eq!( - sess.get_pending_input_with_metadata().await, - vec![ - ( - later, - Some(codex_protocol::models::UserMessageType::PromptQueued), - ), - ( - newer, - Some(codex_protocol::models::UserMessageType::PromptQueued), - ), - ] - ); + assert_eq!(sess.get_pending_input().await, vec![later, newer]); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] diff --git a/codex-rs/core/src/state/mod.rs b/codex-rs/core/src/state/mod.rs index 78e7932a33..642433a786 100644 --- a/codex-rs/core/src/state/mod.rs +++ b/codex-rs/core/src/state/mod.rs @@ -5,6 +5,5 @@ mod turn; pub(crate) use service::SessionServices; pub(crate) use session::SessionState; pub(crate) use turn::ActiveTurn; -pub(crate) use turn::PendingInputItem; pub(crate) use turn::RunningTask; pub(crate) use turn::TaskKind; diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 1aac95ae0a..80576ef4dc 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -10,7 +10,6 @@ use tokio_util::task::AbortOnDropHandle; use codex_protocol::dynamic_tools::DynamicToolResponse; use codex_protocol::models::ResponseInputItem; -use codex_protocol::models::UserMessageType; use codex_protocol::request_permissions::RequestPermissionsResponse; use codex_protocol::request_user_input::RequestUserInputResponse; use codex_rmcp_client::ElicitationResponse; @@ -57,12 +56,6 @@ pub(crate) struct RunningTask { pub(crate) _timer: Option, } -#[derive(Debug, Clone)] -pub(crate) struct PendingInputItem { - pub(crate) input: ResponseInputItem, - pub(crate) user_message_type: Option, -} - impl ActiveTurn { pub(crate) fn add_task(&mut self, task: RunningTask) { let sub_id = task.turn_context.sub_id.clone(); @@ -87,7 +80,7 @@ pub(crate) struct TurnState { pending_user_input: HashMap>, pending_elicitations: HashMap<(String, RequestId), oneshot::Sender>, pending_dynamic_tools: HashMap>, - pending_input: Vec, + pending_input: Vec, granted_permissions: Option, pub(crate) tool_calls: u64, pub(crate) token_usage_at_turn_start: TokenUsage, @@ -182,18 +175,11 @@ impl TurnState { self.pending_dynamic_tools.remove(key) } - pub(crate) fn push_pending_input( - &mut self, - input: ResponseInputItem, - user_message_type: Option, - ) { - self.pending_input.push(PendingInputItem { - input, - user_message_type, - }); + pub(crate) fn push_pending_input(&mut self, input: ResponseInputItem) { + self.pending_input.push(input); } - pub(crate) fn take_pending_input_with_metadata(&mut self) -> Vec { + pub(crate) fn take_pending_input(&mut self) -> Vec { if self.pending_input.is_empty() { Vec::with_capacity(0) } else { @@ -203,7 +189,7 @@ impl TurnState { } } - pub(crate) fn prepend_pending_input(&mut self, mut input: Vec) { + pub(crate) fn prepend_pending_input(&mut self, mut input: Vec) { if input.is_empty() { return; } diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index a1f53d0b49..5888c0b47b 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -252,7 +252,7 @@ impl Session { && at.remove_task(&turn_context.sub_id) { let mut ts = at.turn_state.lock().await; - pending_input = ts.take_pending_input_with_metadata(); + pending_input = ts.take_pending_input(); turn_tool_calls = ts.tool_calls; token_usage_at_turn_start = Some(ts.token_usage_at_turn_start.clone()); should_clear_active_turn = true; @@ -263,7 +263,7 @@ impl Session { drop(active); if !pending_input.is_empty() { for pending_input_item in pending_input { - match inspect_pending_input(self, &turn_context, pending_input_item.input).await { + match inspect_pending_input(self, &turn_context, pending_input_item).await { PendingInputHookDisposition::Accepted(pending_input) => { record_pending_input(self, &turn_context, *pending_input).await; } diff --git a/codex-rs/core/src/tools/js_repl/mod_tests.rs b/codex-rs/core/src/tools/js_repl/mod_tests.rs index 522b8b1a99..33bbd6f031 100644 --- a/codex-rs/core/src/tools/js_repl/mod_tests.rs +++ b/codex-rs/core/src/tools/js_repl/mod_tests.rs @@ -1175,7 +1175,7 @@ console.log(out.type); .await?; assert!(result.output.contains("function_call_output")); assert!(result.content_items.is_empty()); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1241,7 +1241,7 @@ console.log(out.type); }] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1296,7 +1296,7 @@ await codex.emitImage({ bytes: png, mimeType: "image/png" }); }] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1360,7 +1360,7 @@ await codex.emitImage( ] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1415,7 +1415,7 @@ console.log("cell-complete"); }] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1497,7 +1497,7 @@ console.log("helpers-ran"); }, ] ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1541,7 +1541,7 @@ console.log("cell-complete"); .await .expect_err("unawaited invalid emitImage should fail"); assert!(err.to_string().contains("expected non-empty bytes")); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1590,7 +1590,7 @@ console.log("cell-complete"); assert!(result.output.contains("expected non-empty bytes")); assert!(result.output.contains("cell-complete")); assert!(result.content_items.is_empty()); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1637,7 +1637,7 @@ await codex.emitImage({ bytes: png }); .await .expect_err("missing mimeType should fail"); assert!(err.to_string().contains("expected a non-empty mimeType")); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1680,7 +1680,7 @@ await codex.emitImage("https://example.com/image.png"); .await .expect_err("non-data URLs should fail"); assert!(err.to_string().contains("only accepts data URLs")); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1729,7 +1729,7 @@ await codex.emitImage("DATA:image/png;base64,AAA"); }] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1779,7 +1779,7 @@ await codex.emitImage({ bytes: png, mimeType: "image/png", detail: "ultra" }); err.to_string() .contains("only supports detail \"original\"") ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1832,7 +1832,7 @@ await codex.emitImage({ bytes: png, mimeType: "image/png", detail: null }); }] .as_slice() ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -1917,7 +1917,7 @@ await codex.emitImage(out); err.to_string() .contains("does not accept mixed text and image content") ); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } @@ -2079,7 +2079,7 @@ console.log(JSON.stringify(out)); let result = result?; response_watcher_result?; assert!(result.output.contains("hidden-ok")); - assert!(session.get_pending_input_with_metadata().await.is_empty()); + assert!(session.get_pending_input().await.is_empty()); Ok(()) } diff --git a/codex-rs/protocol/src/item_metadata.rs b/codex-rs/protocol/src/item_metadata.rs index 09e04a4938..0860187fcc 100644 --- a/codex-rs/protocol/src/item_metadata.rs +++ b/codex-rs/protocol/src/item_metadata.rs @@ -11,7 +11,7 @@ pub fn stamp_user_message_type_on_input_item(item: &mut ResponseInputItem, kind: } let mut metadata_value = metadata .take() - .unwrap_or_else(|| ResponseItemMessageMetadata::new(None)); + .unwrap_or_else(|| ResponseItemMessageMetadata::new(/*user_message_type*/ None)); metadata_value.user_message_type = Some(kind); *metadata = Some(metadata_value); }