agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: communication pattern v2 (#15647)

Browse Source 
See internal communication
This commit is contained in:
jif-oai
2026-03-24 18:45:49 +00:00
committed by GitHub
Unverified
parent d61c03ca08
commit 773fbf56a4
14 changed files with 714 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/agent/control.rs
+23 -7
View File
@@ -812,24 +812,40 @@ impl AgentControl {
return;
};
let child_thread = state.get_thread(child_thread_id).await.ok();
let message = format_subagent_notification_message(child_reference.as_str(), &status);
if child_agent_path.is_some()
&& child_thread
.as_ref()
.map(|thread| thread.enabled(Feature::MultiAgentV2))
.unwrap_or(true)
{
// Disable v2 completion notifications for now so child turns do
// not enqueue synthetic parent messages on completion.
let Some(child_agent_path) = child_agent_path.clone() else {
return;
};
let Some(parent_agent_path) = child_agent_path
.as_str()
.rsplit_once('/')
.and_then(|(parent, _)| AgentPath::try_from(parent).ok())
else {
return;
};
let communication = InterAgentCommunication::new(
child_agent_path,
parent_agent_path,
Vec::new(),
message,
/*trigger_turn*/ false,
);
let _ = control
.send_inter_agent_communication(parent_thread_id, communication)
.await;
return;
}
let Ok(parent_thread) = state.get_thread(parent_thread_id).await else {
return;
};
parent_thread
.inject_user_message_without_turn(format_subagent_notification_message(
child_reference.as_str(),
&status,
))
.inject_user_message_without_turn(message)
.await;
});
}
@@ -1089,7 +1105,7 @@ fn last_task_message_from_item(item: &ResponseItem) -> Option<String> {
}
match item {
ResponseItem::Message { role, content, .. } if role == "user" => {
ResponseItem::Message { role, .. } if role == "user" => {
let Some(TurnItem::UserMessage(message)) = parse_turn_item(item) else {
return None;
};
codex-rs/core/src/agent/control_tests.rs
+159
View File
@@ -407,6 +407,67 @@ async fn send_input_submits_user_message() {
assert_eq!(captured, Some(expected));
}
#[tokio::test]
async fn send_inter_agent_communication_without_turn_queues_message_without_triggering_turn() {
let harness = AgentControlHarness::new().await;
let (thread_id, thread) = harness.start_thread().await;
let communication = InterAgentCommunication::new(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
Vec::new(),
"hello from tests".to_string(),
false,
);
let submission_id = harness
.control
.send_inter_agent_communication(thread_id, communication.clone())
.await
.expect("send_inter_agent_communication should succeed");
assert!(!submission_id.is_empty());
let expected = (
thread_id,
Op::InterAgentCommunication {
communication: communication.clone(),
},
);
let captured = harness
.manager
.captured_ops()
.into_iter()
.find(|entry| *entry == expected);
assert_eq!(captured, Some(expected));
timeout(Duration::from_secs(5), async {
loop {
if thread
.codex
.session
.has_queued_response_items_for_next_turn()
.await
{
break;
}
sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("inter-agent communication should stay pending");
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(!history_contains_assistant_inter_agent_communication(
&history_items,
&communication
));
}
#[tokio::test]
async fn append_message_records_assistant_message() {
let harness = AgentControlHarness::new().await;
@@ -1057,11 +1118,109 @@ async fn multi_agent_v2_completion_ignores_dead_direct_parent() {
AgentPath::root(),
Vec::new(),
"done".to_string(),
true,
)
));
assert!(!has_subagent_notification(&root_history_items));
}
#[tokio::test]
async fn multi_agent_v2_completion_queues_message_for_direct_parent() {
let harness = AgentControlHarness::new().await;
let (_root_thread_id, root_thread) = harness.start_thread().await;
let (worker_thread_id, _worker_thread) = harness.start_thread().await;
let mut tester_config = harness.config.clone();
let _ = tester_config.features.enable(Feature::MultiAgentV2);
let tester_thread_id = harness
.manager
.start_thread(tester_config)
.await
.expect("tester thread should start")
.thread_id;
let tester_thread = harness
.manager
.get_thread(tester_thread_id)
.await
.expect("tester thread should exist");
let worker_path = AgentPath::root().join("worker_a").expect("worker path");
let tester_path = worker_path.join("tester").expect("tester path");
harness.control.maybe_start_completion_watcher(
tester_thread_id,
Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: worker_thread_id,
depth: 2,
agent_path: Some(tester_path.clone()),
agent_nickname: None,
agent_role: Some("explorer".to_string()),
})),
tester_path.to_string(),
Some(tester_path.clone()),
);
let tester_turn = tester_thread.codex.session.new_default_turn().await;
tester_thread
.codex
.session
.send_event(
tester_turn.as_ref(),
EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: tester_turn.sub_id.clone(),
last_agent_message: Some("done".to_string()),
}),
)
.await;
let expected_message = crate::session_prefix::format_subagent_notification_message(
tester_path.as_str(),
&AgentStatus::Completed(Some("done".to_string())),
);
let expected = (
worker_thread_id,
Op::InterAgentCommunication {
communication: InterAgentCommunication::new(
tester_path.clone(),
worker_path.clone(),
Vec::new(),
expected_message.clone(),
false,
),
},
);
timeout(Duration::from_secs(5), async {
loop {
let captured = harness
.manager
.captured_ops()
.into_iter()
.find(|entry| *entry == expected);
if captured == Some(expected.clone()) {
break;
}
sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("completion watcher should queue a direct-parent message");
let root_history_items = root_thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
assert!(!history_contains_assistant_inter_agent_communication(
&root_history_items,
&InterAgentCommunication::new(
tester_path,
AgentPath::root(),
Vec::new(),
expected_message,
false,
)
));
}
#[tokio::test]
async fn completion_watcher_notifies_parent_when_child_is_missing() {
let harness = AgentControlHarness::new().await;
codex-rs/core/src/codex.rs
+10 -10
View File
@@ -4650,27 +4650,27 @@ mod handlers {
}
}
/// Records an inter-agent assistant envelope and, when requested, wakes the recipient by
/// starting a regular turn if the session is currently idle.
pub async fn inter_agent_communication(
sess: &Arc<Session>,
sub_id: String,
communication: InterAgentCommunication,
) {
let pending_item = communication.to_response_input_item();
if sess
.inject_response_items(vec![pending_item.clone()])
.await
.is_ok()
{
if let Ok(()) = sess.inject_response_items(vec![pending_item.clone()]).await {
return;
}
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref())
.await;
sess.queue_response_items_for_next_turn(vec![pending_item])
.await;
sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new())
.await;
if communication.trigger_turn {
let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
sess.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref())
.await;
sess.spawn_task(turn_context, Vec::new(), crate::tasks::RegularTask::new())
.await;
}
}
pub async fn run_user_shell_command(sess: &Arc<Session>, sub_id: String, command: String) {
codex-rs/core/src/codex/rollout_reconstruction_tests.rs
+1
View File
@@ -41,6 +41,7 @@ fn inter_agent_assistant_message(text: &str) -> ResponseItem {
AgentPath::root().join("worker").unwrap(),
Vec::new(),
text.to_string(),
true,
);
ResponseItem::Message {
id: None,
codex-rs/core/src/context_manager/history_tests.rs
+1
View File
@@ -46,6 +46,7 @@ fn inter_agent_assistant_msg(text: &str) -> ResponseItem {
AgentPath::root().join("worker").unwrap(),
Vec::new(),
text.to_string(),
true,
);
ResponseItem::Message {
id: None,
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+169 -46
View File
@@ -21,8 +21,9 @@ use crate::state::TaskKind;
use crate::tasks::SessionTask;
use crate::tasks::SessionTaskContext;
use crate::tools::context::ToolOutput;
use crate::tools::handlers::multi_agents_v2::AssignTaskHandler as AssignTaskHandlerV2;
use crate::tools::handlers::multi_agents_v2::ListAgentsHandler as ListAgentsHandlerV2;
use crate::tools::handlers::multi_agents_v2::SendInputHandler as SendInputHandlerV2;
use crate::tools::handlers::multi_agents_v2::SendMessageHandler as SendMessageHandlerV2;
use crate::tools::handlers::multi_agents_v2::SpawnAgentHandler as SpawnAgentHandlerV2;
use crate::tools::handlers::multi_agents_v2::WaitAgentHandler as WaitAgentHandlerV2;
use crate::turn_diff_tracker::TurnDiffTracker;
@@ -337,7 +338,7 @@ async fn spawn_agent_errors_when_manager_dropped() {
}
#[tokio::test]
async fn multi_agent_v2_spawn_returns_path_and_send_input_accepts_relative_path() {
async fn multi_agent_v2_spawn_returns_path_and_send_message_accepts_relative_path() {
#[derive(Debug, Deserialize)]
struct SpawnAgentResult {
task_name: String,
@@ -400,18 +401,18 @@ async fn multi_agent_v2_spawn_returns_path_and_send_input_accepts_relative_path(
Some("/root/test_process")
);
SendInputHandlerV2
SendMessageHandlerV2
.handle(invocation(
session.clone(),
turn.clone(),
"send_input",
"send_message",
function_payload(json!({
"target": "test_process",
"message": "continue"
"items": [{"type": "text", "text": "continue"}]
})),
))
.await
.expect("send_input should accept v2 path");
.expect("send_message should accept v2 path");
assert!(manager.captured_ops().iter().any(|(id, op)| {
*id == child_thread_id
@@ -422,6 +423,7 @@ async fn multi_agent_v2_spawn_returns_path_and_send_input_accepts_relative_path(
&& communication.recipient.as_str() == "/root/test_process"
&& communication.other_recipients.is_empty()
&& communication.content == "continue"
&& !communication.trigger_turn
)
}));
}
@@ -647,7 +649,7 @@ async fn multi_agent_v2_list_agents_omits_closed_agents() {
}
#[tokio::test]
async fn multi_agent_v2_send_input_accepts_structured_items() {
async fn multi_agent_v2_send_message_rejects_structured_items() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
@@ -683,7 +685,7 @@ async fn multi_agent_v2_send_input_accepts_structured_items() {
let invocation = invocation(
session,
turn,
"send_input",
"send_message",
function_payload(json!({
"target": agent_id.to_string(),
"items": [
@@ -693,33 +695,19 @@ async fn multi_agent_v2_send_input_accepts_structured_items() {
})),
);
SendInputHandlerV2
.handle(invocation)
.await
.expect("structured items should be accepted in v2");
let expected = Op::UserInput {
items: vec![
UserInput::Mention {
name: "drive".to_string(),
path: "app://google_drive".to_string(),
},
UserInput::Text {
text: "read the folder".to_string(),
text_elements: Vec::new(),
},
],
final_output_json_schema: None,
let Err(err) = SendMessageHandlerV2.handle(invocation).await else {
panic!("structured items should be rejected in v2");
};
let captured = manager
.captured_ops()
.into_iter()
.find(|(id, op)| *id == agent_id && *op == expected);
assert_eq!(captured, Some((agent_id, expected)));
assert_eq!(
err,
FunctionCallError::RespondToModel(
"send_message only supports text content in MultiAgentV2 for now".to_string()
)
);
}
#[tokio::test]
async fn multi_agent_v2_send_input_interrupts_busy_child_without_losing_message() {
async fn multi_agent_v2_send_message_interrupts_busy_child_without_triggering_turn() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
@@ -771,19 +759,161 @@ async fn multi_agent_v2_send_input_interrupts_busy_child_without_losing_message(
)
.await;
SendInputHandlerV2
SendMessageHandlerV2
.handle(invocation(
session,
turn,
"send_input",
session.clone(),
turn.clone(),
"send_message",
function_payload(json!({
"target": agent_id.to_string(),
"message": "continue",
"items": [{"type": "text", "text": "continue"}],
"interrupt": true
})),
))
.await
.expect("interrupting v2 send_input should succeed");
.expect("interrupting v2 send_message should succeed");
let ops = manager.captured_ops();
let ops_for_agent: Vec<&Op> = ops
.iter()
.filter_map(|(id, op)| (*id == agent_id).then_some(op))
.collect();
assert!(ops_for_agent.iter().any(|op| matches!(op, Op::Interrupt)));
assert!(ops_for_agent.iter().any(|op| {
matches!(
op,
Op::InterAgentCommunication { communication }
if communication.author == AgentPath::root()
&& communication.recipient.as_str() == "/root/worker"
&& communication.other_recipients.is_empty()
&& communication.content == "continue"
&& !communication.trigger_turn
)
}));
timeout(Duration::from_secs(5), async {
loop {
if !thread
.codex
.session
.has_queued_response_items_for_next_turn()
.await
{
tokio::time::sleep(Duration::from_millis(10)).await;
continue;
}
let history_items = thread
.codex
.session
.clone_history()
.await
.raw_items()
.to_vec();
let saw_envelope = history_contains_inter_agent_communication(
&history_items,
&InterAgentCommunication::new(
AgentPath::root(),
AgentPath::try_from("/root/worker").expect("agent path"),
Vec::new(),
"continue".to_string(),
false,
),
);
let saw_user_message = history_items.iter().any(|item| {
matches!(
item,
ResponseItem::Message { role, content, .. }
if role == "user"
&& content.iter().any(|content_item| matches!(
content_item,
ContentItem::InputText { text } if text == "continue"
))
)
});
if saw_envelope && !saw_user_message {
panic!("send_message should not materialize the envelope into history");
}
if !saw_user_message {
break;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.expect("interrupting v2 send_message should queue the redirected message without a turn");
let _ = thread
.submit(Op::Shutdown {})
.await
.expect("shutdown should submit");
}
#[tokio::test]
async fn multi_agent_v2_assign_task_interrupts_busy_child_without_losing_message() {
let (mut session, mut turn) = make_session_and_context().await;
let manager = thread_manager();
let root = manager
.start_thread((*turn.config).clone())
.await
.expect("root thread should start");
session.services.agent_control = manager.agent_control();
session.conversation_id = root.thread_id;
let mut config = turn.config.as_ref().clone();
let _ = config.features.enable(Feature::MultiAgentV2);
turn.config = Arc::new(config);
let session = Arc::new(session);
let turn = Arc::new(turn);
SpawnAgentHandlerV2
.handle(invocation(
session.clone(),
turn.clone(),
"spawn_agent",
function_payload(json!({
"message": "boot worker",
"task_name": "worker"
})),
))
.await
.expect("spawn worker");
let agent_id = session
.services
.agent_control
.resolve_agent_reference(session.conversation_id, &turn.session_source, "worker")
.await
.expect("worker should resolve");
let thread = manager
.get_thread(agent_id)
.await
.expect("worker thread should exist");
let active_turn = thread.codex.session.new_default_turn().await;
thread
.codex
.session
.spawn_task(
Arc::clone(&active_turn),
vec![UserInput::Text {
text: "working".to_string(),
text_elements: Vec::new(),
}],
NeverEndingTask,
)
.await;
AssignTaskHandlerV2
.handle(invocation(
session,
turn,
"assign_task",
function_payload(json!({
"target": agent_id.to_string(),
"items": [{"type": "text", "text": "continue"}],
"interrupt": true
})),
))
.await
.expect("interrupting v2 assign_task should succeed");
let ops = manager.captured_ops();
let ops_for_agent: Vec<&Op> = ops
@@ -801,14 +931,6 @@ async fn multi_agent_v2_send_input_interrupts_busy_child_without_losing_message(
&& communication.content == "continue"
)
}));
assert!(!ops_for_agent.iter().any(|op| matches!(
op,
Op::UserInput { items, .. }
if items.iter().any(|item| matches!(
item,
UserInput::Text { text, .. } if text == "continue"
))
)));
timeout(Duration::from_secs(5), async {
loop {
@@ -826,6 +948,7 @@ async fn multi_agent_v2_send_input_interrupts_busy_child_without_losing_message(
AgentPath::try_from("/root/worker").expect("agent path"),
Vec::new(),
"continue".to_string(),
true,
),
);
let saw_user_message = history_items.iter().any(|item| {
@@ -846,7 +969,7 @@ async fn multi_agent_v2_send_input_interrupts_busy_child_without_losing_message(
}
})
.await
.expect("interrupting v2 send_input should preserve the redirected message");
.expect("interrupting v2 assign_task should preserve the redirected message");
let _ = thread
.submit(Op::Shutdown {})
codex-rs/core/src/tools/handlers/multi_agents_v2.rs
+5 -2
View File
@@ -30,12 +30,15 @@ use serde::Deserialize;
use serde::Serialize;
use serde_json::Value as JsonValue;
pub(crate) use assign_task::Handler as AssignTaskHandler;
pub(crate) use list_agents::Handler as ListAgentsHandler;
pub(crate) use send_input::Handler as SendInputHandler;
pub(crate) use send_message::Handler as SendMessageHandler;
pub(crate) use spawn::Handler as SpawnAgentHandler;
pub(crate) use wait::Handler as WaitAgentHandler;
mod assign_task;
mod list_agents;
mod send_input;
mod message_tool;
mod send_message;
mod spawn;
pub(crate) mod wait;
codex-rs/core/src/tools/handlers/multi_agents_v2/assign_task.rs
+23
View File
@@ -0,0 +1,23 @@
use super::message_tool::MessageDeliveryMode;
use super::message_tool::MessageToolResult;
use super::message_tool::handle_message_tool;
use super::*;
pub(crate) struct Handler;
#[async_trait]
impl ToolHandler for Handler {
type Output = MessageToolResult;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
fn matches_kind(&self, payload: &ToolPayload) -> bool {
matches!(payload, ToolPayload::Function { .. })
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
handle_message_tool(invocation, MessageDeliveryMode::TriggerTurn).await
}
}
codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs
+175
View File
@@ -0,0 +1,175 @@
//! Shared argument parsing and dispatch for the v2 text-only agent messaging tools.
//!
//! `send_message` and `assign_task` intentionally expose the same input shape and differ only in
//! whether the resulting `InterAgentCommunication` should wake the target immediately.
use super::*;
use codex_protocol::protocol::InterAgentCommunication;
#[derive(Clone, Copy)]
pub(crate) enum MessageDeliveryMode {
QueueOnly,
TriggerTurn,
}
impl MessageDeliveryMode {
/// Returns the model-visible error message for non-text inputs.
fn unsupported_items_error(self) -> &'static str {
match self {
Self::QueueOnly => "send_message only supports text content in MultiAgentV2 for now",
Self::TriggerTurn => "assign_task only supports text content in MultiAgentV2 for now",
}
}
/// Returns whether the produced communication should start a turn immediately.
fn apply(self, communication: InterAgentCommunication) -> InterAgentCommunication {
match self {
Self::QueueOnly => InterAgentCommunication {
trigger_turn: false,
..communication
},
Self::TriggerTurn => InterAgentCommunication {
trigger_turn: true,
..communication
},
}
}
}
#[derive(Debug, Deserialize)]
/// Input shared by the MultiAgentV2 `send_message` and `assign_task` tools.
pub(crate) struct MessageToolArgs {
pub(crate) target: String,
pub(crate) items: Vec<UserInput>,
#[serde(default)]
pub(crate) interrupt: bool,
}
#[derive(Debug, Serialize)]
/// Tool result shared by the MultiAgentV2 message-delivery tools.
pub(crate) struct MessageToolResult {
submission_id: String,
}
impl ToolOutput for MessageToolResult {
fn log_preview(&self) -> String {
tool_output_json_text(self, "multi_agent_message")
}
fn success_for_logging(&self) -> bool {
true
}
fn to_response_item(&self, call_id: &str, payload: &ToolPayload) -> ResponseInputItem {
tool_output_response_item(call_id, payload, self, Some(true), "multi_agent_message")
}
fn code_mode_result(&self, _payload: &ToolPayload) -> JsonValue {
tool_output_code_mode_result(self, "multi_agent_message")
}
}
/// Validates that the tool input is non-empty text-only content and returns its preview string.
fn text_content(
items: &[UserInput],
mode: MessageDeliveryMode,
) -> Result<String, FunctionCallError> {
if items.is_empty() {
return Err(FunctionCallError::RespondToModel(
"Items can't be empty".to_string(),
));
}
if items
.iter()
.all(|item| matches!(item, UserInput::Text { .. }))
{
return Ok(input_preview(items));
}
Err(FunctionCallError::RespondToModel(
mode.unsupported_items_error().to_string(),
))
}
/// Handles the shared MultiAgentV2 text-message flow for both `send_message` and `assign_task`.
pub(crate) async fn handle_message_tool(
invocation: ToolInvocation,
mode: MessageDeliveryMode,
) -> Result<MessageToolResult, FunctionCallError> {
let ToolInvocation {
session,
turn,
payload,
call_id,
..
} = invocation;
let arguments = function_arguments(payload)?;
let args: MessageToolArgs = parse_arguments(&arguments)?;
let receiver_thread_id = resolve_agent_target(&session, &turn, &args.target).await?;
let prompt = text_content(&args.items, mode)?;
let receiver_agent = session
.services
.agent_control
.get_agent_metadata(receiver_thread_id)
.unwrap_or_default();
if args.interrupt {
session
.services
.agent_control
.interrupt_agent(receiver_thread_id)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err))?;
}
session
.send_event(
&turn,
CollabAgentInteractionBeginEvent {
call_id: call_id.clone(),
sender_thread_id: session.conversation_id,
receiver_thread_id,
prompt: prompt.clone(),
}
.into(),
)
.await;
let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| {
FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string())
})?;
let communication = InterAgentCommunication::new(
turn.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root),
receiver_agent_path,
Vec::new(),
prompt.clone(),
/*trigger_turn*/ true,
);
let result = session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, mode.apply(communication))
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err));
let status = session
.services
.agent_control
.get_status(receiver_thread_id)
.await;
session
.send_event(
&turn,
CollabAgentInteractionEndEvent {
call_id,
sender_thread_id: session.conversation_id,
receiver_thread_id,
receiver_agent_nickname: receiver_agent.agent_nickname,
receiver_agent_role: receiver_agent.agent_role,
prompt,
status,
}
.into(),
)
.await;
let submission_id = result?;
Ok(MessageToolResult { submission_id })
}
codex-rs/core/src/tools/handlers/multi_agents_v2/send_input.rs
-145
View File
@@ -1,145 +0,0 @@
use super::*;
use codex_protocol::protocol::InterAgentCommunication;
pub(crate) struct Handler;
fn can_use_v2_inter_agent_instruction(items: &[UserInput]) -> bool {
items
.iter()
.all(|item| matches!(item, UserInput::Text { .. }))
}
#[async_trait]
impl ToolHandler for Handler {
type Output = SendInputResult;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
fn matches_kind(&self, payload: &ToolPayload) -> bool {
matches!(payload, ToolPayload::Function { .. })
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
let ToolInvocation {
session,
turn,
payload,
call_id,
..
} = invocation;
let arguments = function_arguments(payload)?;
let args: SendInputArgs = parse_arguments(&arguments)?;
let receiver_thread_id = resolve_agent_target(&session, &turn, &args.target).await?;
let input_items = parse_collab_input(args.message, args.items)?;
let prompt = input_preview(&input_items);
let receiver_agent = session
.services
.agent_control
.get_agent_metadata(receiver_thread_id)
.unwrap_or_default();
if args.interrupt {
session
.services
.agent_control
.interrupt_agent(receiver_thread_id)
.await
.map_err(|err| collab_agent_error(receiver_thread_id, err))?;
}
session
.send_event(
&turn,
CollabAgentInteractionBeginEvent {
call_id: call_id.clone(),
sender_thread_id: session.conversation_id,
receiver_thread_id,
prompt: prompt.clone(),
}
.into(),
)
.await;
let result = if can_use_v2_inter_agent_instruction(&input_items) {
let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| {
FunctionCallError::RespondToModel(
"target agent is missing an agent_path".to_string(),
)
})?;
let communication = InterAgentCommunication::new(
turn.session_source
.get_agent_path()
.unwrap_or_else(AgentPath::root),
receiver_agent_path,
Vec::new(),
prompt.clone(),
);
session
.services
.agent_control
.send_inter_agent_communication(receiver_thread_id, communication)
.await
} else {
session
.services
.agent_control
.send_input(receiver_thread_id, input_items)
.await
}
.map_err(|err| collab_agent_error(receiver_thread_id, err));
let status = session
.services
.agent_control
.get_status(receiver_thread_id)
.await;
session
.send_event(
&turn,
CollabAgentInteractionEndEvent {
call_id,
sender_thread_id: session.conversation_id,
receiver_thread_id,
receiver_agent_nickname: receiver_agent.agent_nickname,
receiver_agent_role: receiver_agent.agent_role,
prompt,
status,
}
.into(),
)
.await;
let submission_id = result?;
Ok(SendInputResult { submission_id })
}
}
#[derive(Debug, Deserialize)]
struct SendInputArgs {
target: String,
message: Option<String>,
items: Option<Vec<UserInput>>,
#[serde(default)]
interrupt: bool,
}
#[derive(Debug, Serialize)]
pub(crate) struct SendInputResult {
submission_id: String,
}
impl ToolOutput for SendInputResult {
fn log_preview(&self) -> String {
tool_output_json_text(self, "send_input")
}
fn success_for_logging(&self) -> bool {
true
}
fn to_response_item(&self, call_id: &str, payload: &ToolPayload) -> ResponseInputItem {
tool_output_response_item(call_id, payload, self, Some(true), "send_input")
}
fn code_mode_result(&self, _payload: &ToolPayload) -> JsonValue {
tool_output_code_mode_result(self, "send_input")
}
}
codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs
+23
View File
@@ -0,0 +1,23 @@
use super::message_tool::MessageDeliveryMode;
use super::message_tool::MessageToolResult;
use super::message_tool::handle_message_tool;
use super::*;
pub(crate) struct Handler;
#[async_trait]
impl ToolHandler for Handler {
type Output = MessageToolResult;
fn kind(&self) -> ToolKind {
ToolKind::Function
}
fn matches_kind(&self, payload: &ToolPayload) -> bool {
matches!(payload, ToolPayload::Function { .. })
}
async fn handle(&self, invocation: ToolInvocation) -> Result<Self::Output, FunctionCallError> {
handle_message_tool(invocation, MessageDeliveryMode::QueueOnly).await
}
}
codex-rs/core/src/tools/spec.rs
+91 -3
View File
@@ -1448,6 +1448,80 @@ fn create_send_input_tool() -> ToolSpec {
})
}
fn create_send_message_tool() -> ToolSpec {
let properties = BTreeMap::from([
(
"target".to_string(),
JsonSchema::String {
description: Some(
"Agent id or canonical task name to message (from spawn_agent).".to_string(),
),
},
),
("items".to_string(), create_collab_input_items_schema()),
(
"interrupt".to_string(),
JsonSchema::Boolean {
description: Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message."
.to_string(),
),
},
),
]);
ToolSpec::Function(ResponsesApiTool {
name: "send_message".to_string(),
description: "Add a message to an existing agent without triggering a new turn. Use interrupt=true to stop the current task first. In MultiAgentV2, this tool currently supports text content only."
.to_string(),
strict: false,
defer_loading: None,
parameters: JsonSchema::Object {
properties,
required: Some(vec!["target".to_string(), "items".to_string()]),
additional_properties: Some(false.into()),
},
output_schema: Some(send_input_output_schema()),
})
}
fn create_assign_task_tool() -> ToolSpec {
let properties = BTreeMap::from([
(
"target".to_string(),
JsonSchema::String {
description: Some(
"Agent id or canonical task name to message (from spawn_agent).".to_string(),
),
},
),
("items".to_string(), create_collab_input_items_schema()),
(
"interrupt".to_string(),
JsonSchema::Boolean {
description: Some(
"When true, stop the agent's current task and handle this immediately. When false (default), queue this message."
.to_string(),
),
},
),
]);
ToolSpec::Function(ResponsesApiTool {
name: "assign_task".to_string(),
description: "Add a message to an existing agent and trigger a turn in the target. Use interrupt=true to redirect work immediately. In MultiAgentV2, this tool currently supports text content only."
.to_string(),
strict: false,
defer_loading: None,
parameters: JsonSchema::Object {
properties,
required: Some(vec!["target".to_string(), "items".to_string()]),
additional_properties: Some(false.into()),
},
output_schema: Some(send_input_output_schema()),
})
}
fn create_resume_agent_tool() -> ToolSpec {
let mut properties = BTreeMap::new();
properties.insert(
@@ -2695,8 +2769,9 @@ pub(crate) fn build_specs_with_discoverable_tools(
use crate::tools::handlers::multi_agents::SendInputHandler;
use crate::tools::handlers::multi_agents::SpawnAgentHandler;
use crate::tools::handlers::multi_agents::WaitAgentHandler;
use crate::tools::handlers::multi_agents_v2::AssignTaskHandler as AssignTaskHandlerV2;
use crate::tools::handlers::multi_agents_v2::ListAgentsHandler as ListAgentsHandlerV2;
use crate::tools::handlers::multi_agents_v2::SendInputHandler as SendInputHandlerV2;
use crate::tools::handlers::multi_agents_v2::SendMessageHandler as SendMessageHandlerV2;
use crate::tools::handlers::multi_agents_v2::SpawnAgentHandler as SpawnAgentHandlerV2;
use crate::tools::handlers::multi_agents_v2::WaitAgentHandler as WaitAgentHandlerV2;
use std::sync::Arc;
@@ -3085,10 +3160,22 @@ pub(crate) fn build_specs_with_discoverable_tools(
);
push_tool_spec(
&mut builder,
create_send_input_tool(),
if config.multi_agent_v2 {
create_send_message_tool()
} else {
create_send_input_tool()
},
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
if config.multi_agent_v2 {
push_tool_spec(
&mut builder,
create_assign_task_tool(),
/*supports_parallel_tool_calls*/ false,
config.code_mode_enabled,
);
}
if !config.multi_agent_v2 {
push_tool_spec(
&mut builder,
@@ -3122,7 +3209,8 @@ pub(crate) fn build_specs_with_discoverable_tools(
config.code_mode_enabled,
);
builder.register_handler("spawn_agent", Arc::new(SpawnAgentHandlerV2));
builder.register_handler("send_input", Arc::new(SendInputHandlerV2));
builder.register_handler("send_message", Arc::new(SendMessageHandlerV2));
builder.register_handler("assign_task", Arc::new(AssignTaskHandlerV2));
builder.register_handler("wait_agent", Arc::new(WaitAgentHandlerV2));
builder.register_handler("list_agents", Arc::new(ListAgentsHandlerV2));
} else {
codex-rs/core/src/tools/spec_tests.rs
+31 -6
View File
@@ -550,7 +550,8 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() {
&tools,
&[
"spawn_agent",
"send_input",
"send_message",
"assign_task",
"wait_agent",
"close_agent",
"list_agents",
@@ -584,9 +585,9 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() {
json!(["agent_id", "task_name", "nickname"])
);
let send_input = find_tool(&tools, "send_input");
let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = &send_input.spec else {
panic!("send_input should be a function tool");
let send_message = find_tool(&tools, "send_message");
let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = &send_message.spec else {
panic!("send_message should be a function tool");
};
let JsonSchema::Object {
properties,
@@ -594,10 +595,33 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() {
..
} = parameters
else {
panic!("send_input should use object params");
panic!("send_message should use object params");
};
assert!(properties.contains_key("target"));
assert_eq!(required.as_ref(), Some(&vec!["target".to_string()]));
assert!(!properties.contains_key("message"));
assert_eq!(
required.as_ref(),
Some(&vec!["target".to_string(), "items".to_string()])
);
let assign_task = find_tool(&tools, "assign_task");
let ToolSpec::Function(ResponsesApiTool { parameters, .. }) = &assign_task.spec else {
panic!("assign_task should be a function tool");
};
let JsonSchema::Object {
properties,
required,
..
} = parameters
else {
panic!("assign_task should use object params");
};
assert!(properties.contains_key("target"));
assert!(!properties.contains_key("message"));
assert_eq!(
required.as_ref(),
Some(&vec!["target".to_string(), "items".to_string()])
);
let wait_agent = find_tool(&tools, "wait_agent");
let ToolSpec::Function(ResponsesApiTool {
@@ -652,6 +676,7 @@ fn test_build_specs_multi_agent_v2_uses_task_names_and_hides_resume() {
output_schema["properties"]["agents"]["items"]["required"],
json!(["agent_name", "agent_status", "last_task_message"])
);
assert_lacks_tool_name(&tools, "send_input");
assert_lacks_tool_name(&tools, "resume_agent");
}
codex-rs/protocol/src/protocol.rs
+3
View File
@@ -518,6 +518,7 @@ pub struct InterAgentCommunication {
#[serde(default)]
pub other_recipients: Vec<AgentPath>,
pub content: String,
pub trigger_turn: bool,
}
impl InterAgentCommunication {
@@ -526,12 +527,14 @@ impl InterAgentCommunication {
recipient: AgentPath,
other_recipients: Vec<AgentPath>,
content: String,
trigger_turn: bool,
) -> Self {
Self {
author,
recipient,
other_recipients,
content,
trigger_turn,
}
}