openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

add directly to response input item

Browse Source
This commit is contained in:
Roy Han
2026-03-20 18:03:20 -07:00
parent b8bf118b56
commit 826297a899
7 changed files with 56 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+20 -36
View File
@@ -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<UserMessageType>)> {
pub async fn get_pending_input(&self) -> Vec<ResponseInputItem> {
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<UserMessageType>)>,
input: Vec<ResponseInputItem>,
) -> 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::<Vec<_>>();
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;
codex-rs/core/src/codex_tests.rs
+13 -32
View File
@@ -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)]
codex-rs/core/src/state/mod.rs
-1
View File
@@ -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;
codex-rs/core/src/state/turn.rs
+5 -19
View File
@@ -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<codex_otel::Timer>,
}
#[derive(Debug, Clone)]
pub(crate) struct PendingInputItem {
pub(crate) input: ResponseInputItem,
pub(crate) user_message_type: Option<UserMessageType>,
}
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<String, oneshot::Sender<RequestUserInputResponse>>,
pending_elicitations: HashMap<(String, RequestId), oneshot::Sender<ElicitationResponse>>,
pending_dynamic_tools: HashMap<String, oneshot::Sender<DynamicToolResponse>>,
pending_input: Vec<PendingInputItem>,
pending_input: Vec<ResponseInputItem>,
granted_permissions: Option<PermissionProfile>,
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<UserMessageType>,
) {
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<PendingInputItem> {
pub(crate) fn take_pending_input(&mut self) -> Vec<ResponseInputItem> {
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<PendingInputItem>) {
pub(crate) fn prepend_pending_input(&mut self, mut input: Vec<ResponseInputItem>) {
if input.is_empty() {
return;
}
codex-rs/core/src/tasks/mod.rs
+2 -2
View File
@@ -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;
}
codex-rs/core/src/tools/js_repl/mod_tests.rs
+15 -15
View File
@@ -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(())
}
codex-rs/protocol/src/item_metadata.rs
+1 -1
View File
@@ -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);
}