Add the async user message tool (#39319)

## What changed

- Add `send_user_message_async` for root agents when the async-message feature is enabled and the model advertises tool support.
- Emit the supplied text as an asynchronous agent message and immediately return an accepted result so the turn can continue.
- Keep the user-visible update out of the model's input context.

## Testing

- Add an integration test covering tool registration, item events, continued turn execution, and model context.

GitOrigin-RevId: 13d5168cb8cd6742db17bb74a15c910994e9c98b

codex-rs/core/src/tools/handlers/mod.rs

@@ -23,6 +23,7 @@ mod request_plugin_install;
 pub(crate) mod request_plugin_install_spec;
 mod request_user_input;
 pub(crate) mod request_user_input_spec;
+mod send_user_message_async;
 mod shell;
 pub(crate) mod shell_spec;
 mod sleep;
@@ -68,6 +69,7 @@ pub use plan::PlanHandler;
 pub use request_permissions::RequestPermissionsHandler;
 pub use request_plugin_install::RequestPluginInstallHandler;
 pub use request_user_input::RequestUserInputHandler;
+pub use send_user_message_async::SendUserMessageAsyncHandler;
 pub use shell::ShellCommandHandler;
 pub(crate) use shell::ShellCommandHandlerOptions;
 pub use sleep::SleepHandler;

codex-rs/core/src/tools/handlers/send_user_message_async.rs

@@ -0,0 +1,101 @@
+use crate::function_tool::FunctionCallError;
+use crate::tools::context::FunctionToolOutput;
+use crate::tools::context::ToolInvocation;
+use crate::tools::context::ToolPayload;
+use crate::tools::context::boxed_tool_output;
+use crate::tools::handlers::parse_arguments;
+use crate::tools::registry::CoreToolRuntime;
+use crate::tools::registry::ToolExecutor;
+use codex_protocol::items::AgentMessageContent;
+use codex_protocol::items::AgentMessageDelivery;
+use codex_protocol::items::AgentMessageItem;
+use codex_protocol::items::TurnItem;
+use codex_protocol::models::MessagePhase;
+use codex_tools::JsonSchema;
+use codex_tools::ResponsesApiTool;
+use codex_tools::ToolName;
+use codex_tools::ToolSpec;
+use serde::Deserialize;
+use std::collections::BTreeMap;
+
+const TOOL_NAME: &str = "send_user_message_async";
+
+pub struct SendUserMessageAsyncHandler;
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SendUserMessageAsyncArgs {
+    message: String,
+}
+
+impl ToolExecutor<ToolInvocation> for SendUserMessageAsyncHandler {
+    fn tool_name(&self) -> ToolName {
+        ToolName::plain(TOOL_NAME)
+    }
+
+    fn spec(&self) -> ToolSpec {
+        let properties = BTreeMap::from([(
+            "message".to_string(),
+            JsonSchema::string(Some(
+                "The concise question or update to send to the user.".to_string(),
+            )),
+        )]);
+
+        ToolSpec::Function(ResponsesApiTool {
+            name: TOOL_NAME.to_string(),
+            description: "Send a concise, user-visible acknowledgment, important update, or blocking question. Returns immediately; any reply arrives asynchronously as a new user message."
+                .to_string(),
+            strict: false,
+            defer_loading: None,
+            parameters: JsonSchema::object(
+                properties,
+                Some(vec!["message".to_string()]),
+                /*additional_properties*/ Some(false.into()),
+            ),
+            output_schema: None,
+        })
+    }
+
+    fn handle(&self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'_> {
+        Box::pin(async move {
+            let ToolInvocation {
+                session,
+                turn,
+                call_id,
+                payload,
+                ..
+            } = invocation;
+            let ToolPayload::Function { arguments } = payload else {
+                return Err(FunctionCallError::RespondToModel(format!(
+                    "{TOOL_NAME} handler received unsupported payload"
+                )));
+            };
+            let args: SendUserMessageAsyncArgs = parse_arguments(&arguments)?;
+            let message = args.message.trim();
+            if message.is_empty() {
+                return Err(FunctionCallError::RespondToModel(
+                    "message must not be empty".to_string(),
+                ));
+            }
+
+            let item = TurnItem::AgentMessage(AgentMessageItem {
+                id: call_id,
+                content: vec![AgentMessageContent::Text {
+                    text: message.to_string(),
+                }],
+                phase: Some(MessagePhase::FinalAnswer),
+                memory_citation: None,
+                delivery: Some(AgentMessageDelivery::Async),
+            });
+            session.emit_turn_item_started(turn.as_ref(), &item).await;
+            session.emit_turn_item_completed(turn.as_ref(), item).await;
+
+            Ok(boxed_tool_output(FunctionToolOutput::from_text(
+                r#"{"accepted":true}"#.to_string(),
+                /*success*/ Some(true),
+            )))
+        })
+    }
+}
+
+impl CoreToolRuntime for SendUserMessageAsyncHandler {}

codex-rs/core/src/tools/spec_plan.rs

@@ -24,6 +24,7 @@ use crate::tools::handlers::ReadMcpResourceHandler;
 use crate::tools::handlers::RequestPermissionsHandler;
 use crate::tools::handlers::RequestPluginInstallHandler;
 use crate::tools::handlers::RequestUserInputHandler;
+use crate::tools::handlers::SendUserMessageAsyncHandler;
 use crate::tools::handlers::ShellCommandHandler;
 use crate::tools::handlers::ShellCommandHandlerOptions;
 use crate::tools::handlers::SleepHandler;
@@ -1061,6 +1062,17 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, registry: &mut Tool
         );
     }
 
+    if !turn_context.session_source.is_non_root_agent()
+        && features.enabled(Feature::SendAsyncMessage)
+        && turn_context
+            .model_info
+            .experimental_supported_tools
+            .iter()
+            .any(|tool| tool == "send_user_message_async")
+    {
+        registry.add(SendUserMessageAsyncHandler);
+    }
+
     if environment_mode.has_environment() && features.enabled(Feature::RequestPermissionsTool) {
         registry.add(RequestPermissionsHandler);
     }

codex-rs/core/tests/suite/mod.rs

@@ -149,6 +149,7 @@ mod rollout_list_find;
 mod safety_buffering;
 mod safety_check_downgrade;
 mod search_tool;
+mod send_user_message_async;
 mod shell_command;
 mod shell_serialization;
 mod shell_snapshot;

codex-rs/core/tests/suite/send_user_message_async.rs

@@ -0,0 +1,152 @@
+use anyhow::Result;
+use codex_core::TurnInputRequest;
+use codex_features::Feature;
+use codex_protocol::items::AgentMessageContent;
+use codex_protocol::items::AgentMessageDelivery;
+use codex_protocol::items::AgentMessageItem;
+use codex_protocol::items::TurnItem;
+use codex_protocol::models::MessagePhase;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::user_input::UserInput;
+use core_test_support::responses::ev_assistant_message;
+use core_test_support::responses::ev_completed;
+use core_test_support::responses::ev_function_call_with_namespace;
+use core_test_support::responses::ev_response_created;
+use core_test_support::responses::mount_sse_sequence;
+use core_test_support::responses::sse;
+use core_test_support::responses::start_mock_server;
+use core_test_support::skip_if_no_network;
+use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
+use core_test_support::wait_for_event_match;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn send_user_message_async_emits_item_and_does_not_end_the_turn() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    const CALL_ID: &str = "async-message-call";
+    const MESSAGE: &str = "Still investigating.";
+
+    let server = start_mock_server().await;
+    let responses = mount_sse_sequence(
+        &server,
+        vec![
+            sse(vec![
+                ev_response_created("resp-1"),
+                ev_function_call_with_namespace(
+                    CALL_ID,
+                    "functions",
+                    "send_user_message_async",
+                    &json!({ "message": MESSAGE }).to_string(),
+                ),
+                ev_completed("resp-1"),
+            ]),
+            sse(vec![
+                ev_response_created("resp-2"),
+                ev_assistant_message("final-message", "Finished."),
+                ev_completed("resp-2"),
+            ]),
+        ],
+    )
+    .await;
+    let test = test_codex()
+        .with_model_info_override("gpt-5.2", |model| {
+            model
+                .experimental_supported_tools
+                .push("send_user_message_async".to_string());
+        })
+        .with_config(|config| {
+            config
+                .features
+                .enable(Feature::SendAsyncMessage)
+                .expect("test config should allow enabling async messages");
+        })
+        .build_with_auto_env(&server)
+        .await?;
+
+    test.codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "Keep me updated.".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+
+    let started = wait_for_event_match(test.codex.as_ref(), |event| {
+        let EventMsg::ItemStarted(event) = event else {
+            return None;
+        };
+        let TurnItem::AgentMessage(message) = &event.item else {
+            return None;
+        };
+        if message.delivery != Some(AgentMessageDelivery::Async) {
+            return None;
+        }
+        Some(message.clone())
+    })
+    .await;
+    assert_eq!(
+        serde_json::to_value(&started)?,
+        serde_json::to_value(AgentMessageItem {
+            id: CALL_ID.to_string(),
+            content: vec![AgentMessageContent::Text {
+                text: MESSAGE.to_string(),
+            }],
+            phase: Some(MessagePhase::FinalAnswer),
+            memory_citation: None,
+            delivery: Some(AgentMessageDelivery::Async),
+        })?
+    );
+
+    let completed = wait_for_event_match(test.codex.as_ref(), |event| {
+        let EventMsg::ItemCompleted(event) = event else {
+            return None;
+        };
+        let TurnItem::AgentMessage(message) = &event.item else {
+            return None;
+        };
+        if message.delivery != Some(AgentMessageDelivery::Async) {
+            return None;
+        }
+        Some(message.clone())
+    })
+    .await;
+    assert_eq!(
+        serde_json::to_value(completed)?,
+        serde_json::to_value(started)?
+    );
+
+    wait_for_event(test.codex.as_ref(), |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let requests = responses.requests();
+    assert_eq!(requests.len(), 2);
+    assert!(
+        requests[0].body_json()["tools"]
+            .as_array()
+            .is_some_and(|tools| {
+                tools.iter().any(|tool| {
+                    tool["type"] == "function" && tool["name"] == "send_user_message_async"
+                })
+            }),
+        "the async message tool should be directly visible to the model"
+    );
+    assert_eq!(
+        requests[1].function_call_output_text(CALL_ID),
+        Some(r#"{"accepted":true}"#.to_string())
+    );
+    let has_synthetic_assistant_message = requests[1].input().into_iter().any(|item| {
+        item["type"] == "message"
+            && item["role"] == "assistant"
+            && item.to_string().contains(MESSAGE)
+    });
+    assert!(
+        !has_synthetic_assistant_message,
+        "the user-visible item should not inject a synthetic assistant message into model context"
+    );
+
+    Ok(())
+}