diff --git a/codex-rs/core/src/tools/handlers/mod.rs b/codex-rs/core/src/tools/handlers/mod.rs index 25a2ae1508..4ce3be4a24 100644 --- a/codex-rs/core/src/tools/handlers/mod.rs +++ b/codex-rs/core/src/tools/handlers/mod.rs @@ -24,6 +24,7 @@ pub(crate) mod request_plugin_install_spec; mod request_user_input; mod request_user_input_async; pub(crate) mod request_user_input_spec; +mod send_message_to_user_async; pub(crate) mod shell_spec; mod sleep; mod test_sync; @@ -70,6 +71,7 @@ pub use request_permissions::RequestPermissionsHandler; pub use request_plugin_install::RequestPluginInstallHandler; pub use request_user_input::RequestUserInputHandler; pub use request_user_input_async::RequestUserInputAsyncHandler; +pub use send_message_to_user_async::SendMessageToUserAsyncHandler; pub use sleep::SleepHandler; pub use test_sync::TestSyncHandler; pub(crate) use tool_search::ToolSearchHandlerCache; diff --git a/codex-rs/core/src/tools/handlers/send_message_to_user_async.rs b/codex-rs/core/src/tools/handlers/send_message_to_user_async.rs new file mode 100644 index 0000000000..2dcc61ee93 --- /dev/null +++ b/codex-rs/core/src/tools/handlers/send_message_to_user_async.rs @@ -0,0 +1,109 @@ +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_message_to_user_async"; + +pub struct SendMessageToUserAsyncHandler; + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct SendMessageToUserAsyncArgs { + message: String, +} + +impl ToolExecutor for SendMessageToUserAsyncHandler { + 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 message that needs the user's attention during ongoing work. The tool returns immediately without ending the turn or waiting for a reply; any reply arrives asynchronously as a new user message. Use this tool to ask for missing information, preferences, constraints, clarification, or approval; report a critical blocker or a finding that may change the task's direction; or answer a user question or status request received while work is still in progress. Use this tool when a message needs the user's immediate attention; use commentary for routine progress and intermediate context. Use clear formatting, such as bolding questions, to make requests easy to notice and answer." + .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<'a>(&'a self, invocation: ToolInvocation) -> codex_tools::ToolExecutorFuture<'a> + where + ToolInvocation: 'a, + { + 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: SendMessageToUserAsyncArgs = 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), + questions: None, + }); + 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 SendMessageToUserAsyncHandler { + fn is_builtin_control_tool(&self) -> bool { + true + } +} diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 908bb143af..ef31ab2034 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -25,6 +25,7 @@ use crate::tools::handlers::RequestPermissionsHandler; use crate::tools::handlers::RequestPluginInstallHandler; use crate::tools::handlers::RequestUserInputAsyncHandler; use crate::tools::handlers::RequestUserInputHandler; +use crate::tools::handlers::SendMessageToUserAsyncHandler; use crate::tools::handlers::SleepHandler; use crate::tools::handlers::TestSyncHandler; use crate::tools::handlers::ToolSearchHandlerCache; @@ -1183,6 +1184,16 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, registry: &mut Tool ); } + if !turn_context.session_source.is_non_root_agent() + && context + .model_info + .experimental_supported_tools + .iter() + .any(|tool| tool == "send_message_to_user_async") + { + registry.add_with_exposure(SendMessageToUserAsyncHandler, ToolExposure::DirectModelOnly); + } + if environment_mode.has_environment() && features.enabled(Feature::RequestPermissionsTool) { registry.add(RequestPermissionsHandler); } diff --git a/codex-rs/core/tests/suite/request_user_input_async.rs b/codex-rs/core/tests/suite/request_user_input_async.rs index fdb6cb93fb..7e20abf110 100644 --- a/codex-rs/core/tests/suite/request_user_input_async.rs +++ b/codex-rs/core/tests/suite/request_user_input_async.rs @@ -120,6 +120,195 @@ async fn persistent_async_message_guidance_follows_tool_availability( Ok(()) } +#[test_case(SessionSource::Exec, None, false; "root_without_opt_in")] +#[test_case(SessionSource::Exec, Some("send_user_message_async"), false; "root_with_legacy_question_tool")] +#[test_case(SessionSource::Exec, Some("request_user_input_async"), false; "root_with_question_tool")] +#[test_case(SessionSource::Exec, Some("send_message_to_user_async"), true; "root_with_freeform_tool")] +#[test_case(SessionSource::SubAgent(SubAgentSource::Other("test".to_string())), Some("send_message_to_user_async"), false; "subagent_with_freeform_tool")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn freeform_async_message_requires_root_and_exact_catalog_opt_in( + session_source: SessionSource, + catalog_tool: Option<&'static str>, + expected: bool, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let responses = mount_sse_once( + &server, + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + ) + .await; + let test = test_codex() + .with_model_info_override("gpt-5.2", move |model| { + model.tool_mode = Some(ToolMode::CodeModeOnly); + model.experimental_supported_tools = + catalog_tool.map(str::to_string).into_iter().collect(); + }) + .build_with_auto_env(&server) + .await?; + let thread = test + .thread_manager + .start_thread(StartThreadOptions { + session_source: Some(session_source), + ..StartThreadOptions::new(test.config.clone()) + }) + .await? + .thread; + thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Report progress.".to_string(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + + let body = responses.single_request().body_json(); + let tools = body["tools"].as_array().expect("request tools"); + assert_eq!( + tools + .iter() + .any(|tool| tool["name"] == "send_message_to_user_async"), + expected, + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn freeform_async_message_emits_an_item_without_ending_the_turn() -> Result<()> { + skip_if_no_network!(Ok(())); + + const CALL_ID: &str = "freeform-message"; + const MESSAGE: &str = "I found a blocker that changes the plan."; + 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_message_to_user_async", + &json!({ "message": format!(" {MESSAGE} ") }).to_string(), + ), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_function_call_with_namespace( + "blank-message", + "functions", + "send_message_to_user_async", + &json!({ "message": " " }).to_string(), + ), + ev_completed("resp-2"), + ]), + sse(vec![ + ev_response_created("resp-3"), + ev_assistant_message("final-message", "Finished."), + ev_completed("resp-3"), + ]), + ], + ) + .await; + let test = test_codex() + .with_model_info_override("gpt-5.2", |model| { + model.tool_mode = Some(ToolMode::CodeModeOnly); + model.experimental_supported_tools.extend([ + "send_message_to_user_async".to_string(), + "request_user_input_async".to_string(), + ]); + model + .model_messages + .as_mut() + .expect("model instruction metadata") + .tools = Some(ToolMessages { + send_user_message_async: Some(ToolMessage { + description: Some("Questions only.".to_string()), + }), + }); + }) + .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| match event { + EventMsg::ItemStarted(event) if event.item.id() == CALL_ID => Some(event.item.clone()), + _ => None, + }) + .await; + let completed = wait_for_event_match(test.codex.as_ref(), |event| match event { + EventMsg::ItemCompleted(event) if event.item.id() == CALL_ID => Some(event.item.clone()), + _ => None, + }) + .await; + let expected = serde_json::to_value(TurnItem::AgentMessage(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), + questions: None, + }))?; + assert_eq!(serde_json::to_value(started)?, expected); + assert_eq!(serde_json::to_value(completed)?, expected); + wait_for_event(test.codex.as_ref(), |event| { + let item = match event { + EventMsg::ItemStarted(event) => Some(&event.item), + EventMsg::ItemCompleted(event) => Some(&event.item), + _ => None, + }; + assert!(!matches!(item, Some(TurnItem::AgentMessage(message)) if message.delivery == Some(AgentMessageDelivery::Async))); + matches!(event, EventMsg::TurnComplete(_)) + }).await; + + let requests = responses.requests(); + assert_eq!(requests.len(), 3); + let body = requests[0].body_json(); + let tools = body["tools"].as_array().expect("request tools"); + let freeform_tool = tools + .iter() + .find(|tool| tool["name"] == "send_message_to_user_async") + .expect("freeform tool is directly available"); + assert_eq!(freeform_tool["parameters"]["required"], json!(["message"])); + assert!( + freeform_tool["description"] + .as_str() + .expect("description") + .contains("report a critical blocker") + ); + let question_tool = tools + .iter() + .find(|tool| tool["name"] == "request_user_input_async") + .expect("question tool is still directly available"); + assert_eq!(question_tool["description"], "Questions only."); + assert_eq!( + requests[1].function_call_output_text(CALL_ID), + Some(r#"{"accepted":true}"#.to_string()) + ); + assert_eq!( + requests[2].function_call_output_text("blank-message"), + Some("message must not be empty".to_string()) + ); + assert!( + !requests[1] + .input() + .into_iter() + .any(|item| item["type"] == "message" + && item["role"] == "assistant" + && item.to_string().contains(MESSAGE)) + ); + Ok(()) +} + #[test_case(None, "send_user_message_async"; "fallback_description")] #[test_case(None, "request_user_input_async"; "current_catalog_name")] #[test_case(Some(ToolMessages { send_user_message_async: None }), "send_user_message_async"; "missing_tool")]