diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 6861844855..43b63c512b 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/codex-rs/core/src/codex_tests.rs @@ -3795,6 +3795,75 @@ async fn task_finish_emits_turn_item_lifecycle_for_leftover_pending_user_input() )); } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn task_finish_emits_prompt_queued_metadata_for_injected_user_input_when_feature_enabled() { + let (mut sess, tc, rx) = make_session_and_context_with_rx().await; + Arc::get_mut(&mut sess) + .expect("session should be uniquely owned in this test") + .features + .enable(crate::features::Feature::UserMessageTypeMetadata) + .expect("feature flag should be enabled for this test"); + + let input = vec![UserInput::Text { + text: "hello".to_string(), + text_elements: Vec::new(), + }]; + sess.spawn_task( + Arc::clone(&tc), + input, + NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: false, + }, + ) + .await; + + while rx.try_recv().is_ok() {} + + sess.inject_response_items(vec![ResponseInputItem::Message { + role: "user".to_string(), + content: vec![ContentItem::InputText { + text: "late queued pending input".to_string(), + }], + }]) + .await + .expect("inject pending input into active turn"); + + sess.on_task_finished(Arc::clone(&tc), None).await; + + let first = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("expected raw response item event") + .expect("channel open"); + assert!(matches!(first.msg, EventMsg::RawResponseItem(_))); + + let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("expected item started event") + .expect("channel open"); + assert!(matches!( + second.msg, + EventMsg::ItemStarted(ItemStartedEvent { + item: TurnItem::UserMessage(UserMessageItem { metadata: Some(metadata), .. }), + .. + }) if metadata.user_message_type + == Some(codex_protocol::items::UserMessageType::PromptQueued) + )); + + let third = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("expected item completed event") + .expect("channel open"); + assert!(matches!( + third.msg, + EventMsg::ItemCompleted(ItemCompletedEvent { + item: TurnItem::UserMessage(UserMessageItem { metadata: Some(metadata), .. }), + .. + }) if metadata.user_message_type + == Some(codex_protocol::items::UserMessageType::PromptQueued) + )); +} + #[tokio::test] async fn steer_input_requires_active_turn() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 57653400a2..7a243f9f74 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -88,17 +88,6 @@ impl CodexThread { self.codex.steer_input(input, expected_turn_id).await } - /// Use sparingly: inject response input into the active turn. - /// - /// Returns the input unchanged when there is no active turn. - #[doc(hidden)] - pub async fn inject_response_items( - &self, - input: Vec, - ) -> Result<(), Vec> { - self.codex.inject_response_items(input).await - } - pub async fn set_app_server_client_name( &self, app_server_client_name: Option, diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index f126f7f948..e39d4672c0 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -361,137 +361,6 @@ async fn user_message_type_prompt_steering_metadata_is_emitted_when_feature_enab Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn user_message_type_prompt_queued_metadata_is_emitted_when_feature_enabled() --> anyhow::Result<()> { - skip_if_no_network!(Ok(())); - - let server = start_mock_server().await; - let temp = tempdir()?; - let unblock_path = temp.path().join("unblock-queued"); - let command = format!( - "while [ ! -f \"{}\" ]; do sleep 0.01; done; echo done", - unblock_path.display() - ); - let call_id = "shell-queued-call"; - - let responses = mount_sse_sequence( - &server, - vec![ - sse(vec![ - ev_response_created("resp-1"), - ev_function_call( - call_id, - "shell", - &serde_json::to_string(&json!({ - "command": ["/bin/sh", "-c", command], - }))?, - ), - ev_completed("resp-1"), - ]), - sse(vec![ - ev_assistant_message("msg-2", "done"), - ev_completed("resp-2"), - ]), - ], - ) - .await; - - let test = test_codex() - .with_model("gpt-5") - .with_config(|config| { - config - .features - .enable(Feature::UserMessageTypeMetadata) - .expect("feature flag should be enabled for this test"); - }) - .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 queued flow".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - cwd: test.cwd_path().to_path_buf(), - approval_policy: AskForApproval::Never, - 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 queued_text = "queued metadata check"; - assert!( - codex - .inject_response_items(vec![codex_protocol::models::ResponseInputItem::Message { - role: "user".into(), - content: vec![codex_protocol::models::ContentItem::InputText { - text: queued_text.into(), - }], - }]) - .await - .is_ok(), - "inject_response_items should succeed on active turn" - ); - - std::fs::write(&unblock_path, "go")?; - - wait_for_event(&codex, |ev| match ev { - EventMsg::TurnComplete(event) => event.turn_id == turn_id, - _ => false, - }) - .await; - - let deadline = Instant::now() + Duration::from_secs(3); - while responses.requests().len() < 2 { - if Instant::now() >= deadline { - panic!("timed out waiting for second responses request"); - } - sleep(Duration::from_millis(10)).await; - } - - let requests = responses.requests(); - let second_body = requests - .get(1) - .expect("second responses request should be present") - .body_json(); - let input = second_body - .get("input") - .and_then(Value::as_array) - .expect("request input array"); - let queued_message = user_message_item_by_text(input, queued_text); - assert_eq!( - queued_message - .get("metadata") - .and_then(|metadata| metadata.get("user_message_type")) - .and_then(Value::as_str), - Some("prompt_queued") - ); - - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn assistant_message_item_is_emitted() -> anyhow::Result<()> { skip_if_no_network!(Ok(()));