diff --git a/codex-rs/core/src/tools/handlers/mod.rs b/codex-rs/core/src/tools/handlers/mod.rs index 47bcfe0970..07f192013b 100644 --- a/codex-rs/core/src/tools/handlers/mod.rs +++ b/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; diff --git a/codex-rs/core/src/tools/handlers/send_user_message_async.rs b/codex-rs/core/src/tools/handlers/send_user_message_async.rs new file mode 100644 index 0000000000..b6636b0b8f --- /dev/null +++ b/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 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 {} diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 3da01c0023..3de1dfd0ca 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/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); } diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 8c5f742bc9..ee7a971fd9 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/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; diff --git a/codex-rs/core/tests/suite/send_user_message_async.rs b/codex-rs/core/tests/suite/send_user_message_async.rs new file mode 100644 index 0000000000..530e067c08 --- /dev/null +++ b/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(()) +}