inject cleanup

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;

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<ResponseInputItem>,
-    ) -> Result<(), Vec<ResponseInputItem>> {
-        self.codex.inject_response_items(input).await
-    }
-
     pub async fn set_app_server_client_name(
         &self,
         app_server_client_name: Option<String>,

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(()));