diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 21d31dd791..923f9f5628 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -24,6 +24,9 @@ use tokio_util::task::AbortOnDropHandle; pub struct UserInputMetadata { #[serde(default, skip_serializing_if = "Option::is_none")] pub acceptance_order: Option, + /// Optional host-assigned UUID for this particular Responses user item. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub response_item_uuid: Option, #[serde( default, skip_serializing_if = "codex_history::UserInputOrigin::is_user" diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d156184403..2b50e0d565 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -5481,6 +5481,9 @@ impl Session { input.to_vec(), &mut user_image_content_indices, ); + if let Some(uuid) = metadata.response_item_uuid { + response_item.set_id(Some(ResponseItemId::for_admitted_user_input(uuid))); + } if metadata.origin == codex_history::UserInputOrigin::Heartbeat && let ResponseItem::Message { content, diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 0364e6af32..4fa1c9de13 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -13,6 +13,7 @@ //! Sender evidence is read without the active-turn lock and bound to the accepted turn. use super::TurnInput; +use super::UserInputMetadata; use super::session::Session; use super::session::SessionConfiguration; use super::session::SessionSettingsUpdate; @@ -333,6 +334,14 @@ pub(super) async fn handle_recovery( result } +fn user_input_metadata(request: &TurnInputRequest) -> UserInputMetadata { + UserInputMetadata { + origin: UserInputOrigin::from_turn_trigger(request.start.turn_trigger.as_deref()), + response_item_uuid: request.user_item_uuid, + ..Default::default() + } +} + async fn start_or_steer( session: &Arc, request: WithTurnExtensionData, @@ -342,6 +351,7 @@ async fn start_or_steer( request, turn_extension_init, } = request; + let input_metadata = user_input_metadata(&request); let TurnInputRequest { mut input, thread_settings, @@ -350,7 +360,6 @@ async fn start_or_steer( responsesapi_client_metadata, .. } = request; - let origin = UserInputOrigin::from_turn_trigger(start.turn_trigger.as_deref()); let has_explicit_input = match &input { SubmittedTurnInput::UserInput { content, .. } => !content.is_empty(), SubmittedTurnInput::ResponseItem(ResponseItem::FunctionCallOutput { @@ -386,7 +395,7 @@ async fn start_or_steer( /*expected_turn_id*/ None, settings.required_active_final_output_json_schema(), responsesapi_client_metadata.clone(), - origin, + input_metadata, ) .await { @@ -440,7 +449,7 @@ async fn start_or_steer( input, sender_messages.as_ref(), &turn_context.sub_id, - origin, + input_metadata, ) .await, ); @@ -472,6 +481,7 @@ async fn start_if_idle( request, turn_extension_init, } = request; + let input_metadata = user_input_metadata(&request); let TurnInputRequest { mut input, thread_settings, @@ -480,7 +490,6 @@ async fn start_if_idle( responsesapi_client_metadata, .. } = request; - let origin = UserInputOrigin::from_turn_trigger(start.turn_trigger.as_deref()); if session.input_queue.has_trigger_turn_mailbox_items().await { return Ok(TurnInputSubmission::NotSubmitted { reason: NotSubmittedReason::PendingTriggerTurn, @@ -593,7 +602,7 @@ async fn start_if_idle( input, sender_messages.as_ref(), &turn_context.sub_id, - origin, + input_metadata, ) .await, ); @@ -611,7 +620,7 @@ async fn start_if_idle( input, sender_messages.as_ref(), &turn_context.sub_id, - origin, + input_metadata, ) .await, ], @@ -639,6 +648,7 @@ async fn steer( request, turn_extension_init, } = request; + let input_metadata = user_input_metadata(&request); let TurnInputRequest { mut input, thread_settings, @@ -647,7 +657,6 @@ async fn steer( responsesapi_client_metadata, .. } = request; - let origin = UserInputOrigin::from_turn_trigger(start.turn_trigger.as_deref()); if !matches!(&input, SubmittedTurnInput::UserInput { .. }) { return Err(CodexErr::InvalidRequest( "only user input can steer a turn".to_string(), @@ -670,7 +679,7 @@ async fn steer( Some(expected_turn_id.as_str()), settings.required_active_final_output_json_schema(), responsesapi_client_metadata, - origin, + input_metadata, ) .await { @@ -855,7 +864,7 @@ impl Session { expected_turn_id: Option<&str>, required_final_output_json_schema: Option<&Value>, responsesapi_client_metadata: ResponsesApiClientMetadataUpdate, - origin: UserInputOrigin, + input_metadata: UserInputMetadata, ) -> Result<(String, String), NotSubmittedReason> { let mut active = self.active_turn.lock().await; let Some(active_turn) = active.as_mut() else { @@ -914,12 +923,18 @@ impl Session { client_id: client_id.clone(), metadata: super::UserInputMetadata { acceptance_order: Some(self.reserve_user_input_order().await), - origin, + ..input_metadata }, }, input => { - pending_turn_input(self, input.clone(), sender_messages, active_turn_id, origin) - .await + pending_turn_input( + self, + input.clone(), + sender_messages, + active_turn_id, + input_metadata, + ) + .await } }; pending_input.push(input); @@ -981,7 +996,7 @@ async fn pending_turn_input( input: SubmittedTurnInput, sender_messages: Option<&GuardianSenderMessages>, turn_id: &str, - origin: UserInputOrigin, + input_metadata: UserInputMetadata, ) -> TurnInput { let retain = match &input { SubmittedTurnInput::AnnotatedResponseItem(input) => input.annotations.retain, @@ -993,7 +1008,7 @@ async fn pending_turn_input( client_id, metadata: super::UserInputMetadata { acceptance_order: Some(session.reserve_user_input_order().await), - origin, + ..input_metadata }, }, SubmittedTurnInput::ResponseItem(item) diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index 2dca090537..c8df7c18d2 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -1133,8 +1133,108 @@ async fn steer_preserves_request_origin( metadata: crate::session::UserInputMetadata { acceptance_order: Some(0), origin, + ..Default::default() }, }] ); session.abort_all_tasks(TurnAbortReason::Interrupted).await; } + +#[tokio::test] +async fn scored_prompt_keeps_its_submission_item_id() { + let (session, turn, events) = make_session_and_context_with_rx().await; + session + .spawn_task( + Arc::clone(&turn), + Vec::new(), + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + // A legacy steer can present the same UI client ID as an admitted steer. + for id in [Some(first), None, Some(second)] { + let mut request = TurnInputRequest::new(SubmittedTurnInput::UserInput { + content: vec![UserInput::Text { + text: "identical input".into(), + text_elements: vec![], + }], + client_id: Some("same-client-id".into()), + }); + request.user_item_uuid = id; + handle( + &session, + request, + TurnInputMode::Steer { + expected_turn_id: turn.sub_id.clone(), + }, + Uuid::now_v7().to_string(), + ) + .await + .unwrap(); + } + let (pending, _) = session + .input_queue + .get_pending_input(&session.active_turn) + .await; + for input in pending { + let TurnInput::UserInput { + content, + client_id, + metadata, + } = input + else { + panic!("expected admitted user input"); + }; + session + .record_user_prompt_and_emit_turn_item( + &turn, + turn.model_info(), + &content, + client_id, + metadata, + codex_thread_store::PersistContext::Standard, + ) + .await; + } + let history = session.clone_history().await; + let ids: Vec<_> = history + .raw_items() + .filter_map(|item| { + if let ResponseItem::Message { role, id, .. } = item + && role == "user" + { + id.as_ref().map(ToString::to_string) + } else { + None + } + }) + .collect(); + assert_eq!(ids.len(), 3); + assert_eq!(ids[0], format!("msg_input_{first}")); + assert!(!ids[1].starts_with("msg_input_")); + assert_eq!(ids[2], format!("msg_input_{second}")); + let ui_ids = std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event.msg { + codex_protocol::protocol::EventMsg::ItemCompleted(event) => { + let item = codex_app_server_protocol::ThreadItemPayload::from(event.item); + if let codex_app_server_protocol::ThreadItemPayload::UserMessage { + id, + client_id, + .. + } = item + { + assert_eq!(client_id.as_deref(), Some("same-client-id")); + return Some(id); + } + None + } + _ => None, + }) + .collect::>(); + assert_eq!(ui_ids.len(), 3); + session.abort_all_tasks(TurnAbortReason::Interrupted).await; +} diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index e0fe6da767..e9c7f39805 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -322,6 +322,62 @@ fn configure_scenario_catalog(config: &mut Config) { config.cloud_skill_enabled = false; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn admitted_prompt_id_reaches_responses() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let test = test_codex() + .with_config(|config| { + configure_scenario_catalog(config); + config.include_environment_context = false; + }) + .build_with_auto_env(&server) + .await?; + let mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_completed("first")]), + sse(vec![ev_completed("second")]), + ], + ) + .await; + let uuid = uuid::Uuid::new_v4(); + for id in [Some(uuid), None] { + let mut request = TurnInputRequest::user_input(vec![text("Repeat this input.")]); + request.user_item_uuid = id; + test.codex.start_or_steer_turn(request).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + } + + let requests = mock.requests(); + assert_eq!(requests.len(), 2); + let reserved = codex_protocol::ResponseItemId::for_admitted_user_input(uuid); + for (index, request) in requests.iter().enumerate() { + let input = request.input(); + let prompts: Vec<_> = input + .iter() + .filter(|item| { + item["role"] == "user" && item["content"][0]["text"] == "Repeat this input." + }) + .collect(); + assert_eq!(prompts.len(), index + 1); + assert_eq!(prompts[0]["id"], reserved.as_str()); + if let Some(legacy) = prompts.get(1) { + let id = legacy["id"].as_str().expect("legacy message id"); + assert!(id.starts_with("msg_") && !id.starts_with("msg_input_")); + } + } + insta::assert_snapshot!(context_snapshot::format_request_history_snapshot( + "An admitted prompt and an ordinary prompt with identical text reach Responses on successive turns.", + &requests, + &ContextSnapshotOptions::default().rewrite_known_segments(), + )); + Ok(()) +} + #[test_case::test_case(false; "messages")] #[test_case::test_case(true; "tool_outputs")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__admitted_prompt_id_reaches_responses.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__admitted_prompt_id_reaches_responses.snap new file mode 100644 index 0000000000..f099672ffa --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__admitted_prompt_id_reaches_responses.snap @@ -0,0 +1,17 @@ +--- +source: core/tests/suite/scenarios.rs +expression: "context_snapshot::format_request_history_snapshot(\"An admitted prompt and an ordinary prompt with identical text reach Responses on successive turns.\",\n&requests, &ContextSnapshotOptions::default().rewrite_known_segments(),)" +--- +Scenario: An admitted prompt and an ordinary prompt with identical text reach Responses on successive turns. + +## Window 1 +-- request 1 (turn) -- +00:message/developer: + +01:message/developer: + +02:message/user: + Repeat this input. +-- request 2 (turn) -- +03:message/user: + Repeat this input. diff --git a/codex-rs/protocol/src/response_item_id.rs b/codex-rs/protocol/src/response_item_id.rs index 9aa66b78da..24a18425a0 100644 --- a/codex-rs/protocol/src/response_item_id.rs +++ b/codex-rs/protocol/src/response_item_id.rs @@ -25,6 +25,16 @@ impl ResponseItemId { Self(format!("{prefix}_{suffix}")) } + /// Reserved for physical user inputs admitted by the host. Raw item + /// ingestion must reject this namespace when it accepts caller-supplied IDs. + pub fn for_admitted_user_input(uuid: uuid::Uuid) -> Self { + Self::with_suffix("msg_input", uuid) + } + + pub fn is_admitted_user_input(&self) -> bool { + self.0.starts_with("msg_input_") + } + pub fn from_server(value: String) -> Self { Self(value) } diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index c1998070ea..514578d98b 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -59,6 +59,9 @@ pub struct ResponseItemAnnotations { #[derive(Clone, Debug)] pub struct TurnInputRequest { pub input: TurnInput, + /// Optional host-assigned UUID for the Responses message created from [`TurnInput::UserInput`]. + /// Ignored for all other inputs, including raw user-role Responses items. + pub user_item_uuid: Option, pub thread_settings: ThreadSettingsOverrides, pub start: TurnStartOptions, pub additional_context: BTreeMap, @@ -116,6 +119,7 @@ impl TurnInputRequest { pub fn new(input: TurnInput) -> Self { Self { input, + user_item_uuid: None, thread_settings: ThreadSettingsOverrides::default(), start: TurnStartOptions::default(), additional_context: BTreeMap::new(),