diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 6435d93409..926afbe79e 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -67,10 +67,14 @@ use tokio::sync::watch; use tracing::warn; use uuid::Uuid; +pub(crate) use self::delivery::AgentMessage; +pub(crate) use self::delivery::MessageDeliveryError; +pub(crate) use self::delivery::MessageDeliveryMode; pub(crate) use self::execution::AgentExecutionGuard; use self::execution::AgentExecutionLimiter; use self::residency::V2Residency; +mod delivery; mod execution; mod legacy; mod residency; diff --git a/codex-rs/core/src/agent/control/delivery.rs b/codex-rs/core/src/agent/control/delivery.rs new file mode 100644 index 0000000000..034fa0adec --- /dev/null +++ b/codex-rs/core/src/agent/control/delivery.rs @@ -0,0 +1,136 @@ +//! Delivers V2 messages without exposing local loading and eviction to callers. +//! +//! Target checks precede reload, and queue-only messages retain their non-waking semantics. + +use super::AgentControl; +use crate::TurnStartOptions; +use crate::agent::child_config::build_agent_resume_config; +use crate::agent_communication::AgentCommunicationContext; +use crate::agent_communication::AgentCommunicationKind; +use crate::context::ContextualUserFragment; +use crate::context::InterAgentMessage; +use crate::context::InterAgentMessageType; +use crate::session::turn_context::TurnContext; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::protocol::InterAgentCommunication; + +#[derive(Clone, Copy, PartialEq, Eq)] +pub(crate) enum MessageDeliveryMode { + QueueOnly, + TriggerTurn, +} + +/// Keeps model-provided encrypted content distinct from text that needs a context wrapper. +pub(crate) enum AgentMessage { + Plaintext(String), + Encrypted(String), +} + +impl AgentMessage { + pub(crate) fn into_communication( + self, + author: AgentPath, + recipient: AgentPath, + mode: MessageDeliveryMode, + ) -> InterAgentCommunication { + let trigger_turn = mode == MessageDeliveryMode::TriggerTurn; + match self { + Self::Encrypted(message) => InterAgentCommunication::new_encrypted( + author, + recipient, + Vec::new(), + message, + trigger_turn, + ), + Self::Plaintext(message) => { + let message_type = match mode { + MessageDeliveryMode::QueueOnly => InterAgentMessageType::Message, + MessageDeliveryMode::TriggerTurn => InterAgentMessageType::NewTask, + }; + let content = InterAgentMessage::new( + message_type, + recipient.clone(), + author.clone(), + message, + ) + .render(); + InterAgentCommunication::new(author, recipient, Vec::new(), content, trigger_turn) + } + } + } +} + +/// Separates request validation from agent runtime failures so adapters retain their error text. +#[derive(Debug)] +pub(crate) enum MessageDeliveryError { + InvalidRequest(String), + Agent(CodexErr), +} + +impl AgentControl { + /// Checks and delivers to a resolved target, restoring an evicted runtime when necessary. + /// + /// The caller resolves tool-facing names separately so it can attribute failures and + /// interruptions to the target before delivery starts. + pub(crate) async fn deliver_message( + &self, + caller: ThreadId, + turn: &TurnContext, + target: ThreadId, + message: AgentMessage, + mode: MessageDeliveryMode, + ) -> Result { + let receiver_agent = self + .ensure_agent_known(target) + .map_err(MessageDeliveryError::Agent)?; + if mode == MessageDeliveryMode::TriggerTurn + && receiver_agent + .agent_path + .as_ref() + .is_some_and(AgentPath::is_root) + { + return Err(MessageDeliveryError::InvalidRequest( + "Follow-up tasks can't target the root agent".to_string(), + )); + } + let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| { + MessageDeliveryError::InvalidRequest( + "target agent is missing an agent_path".to_string(), + ) + })?; + let resume_config = + build_agent_resume_config(turn).map_err(MessageDeliveryError::InvalidRequest)?; + self.ensure_v2_agent_loaded(resume_config, target, /*parent*/ None) + .await + .map_err(MessageDeliveryError::Agent)?; + let author = turn + .session_source + .get_agent_path() + .unwrap_or_else(AgentPath::root); + let communication = message.into_communication(author, receiver_agent_path.clone(), mode); + let kind = match mode { + MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message, + MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup, + }; + let context = AgentCommunicationContext::new(kind, caller); + let parent_turn_id = + matches!(mode, MessageDeliveryMode::TriggerTurn).then(|| turn.sub_id.clone()); + self.send_inter_agent_communication( + target, + communication, + context, + TurnStartOptions { + parent_turn_id, + root_turn_id: turn.turn_metadata_state.root_turn_id(), + turn_trigger: turn.turn_metadata_state.current_turn_trigger(), + cyber_access_program: turn.cyber_access_program, + ..Default::default() + }, + ) + .await + .map_err(MessageDeliveryError::Agent)?; + Ok(receiver_agent_path) + } +} diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs index b82bdd6a0b..9f4176617f 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2.rs @@ -2,9 +2,7 @@ use crate::agent::AgentStatus; use crate::agent::agent_resolver::resolve_agent_target; -use crate::context::ContextualUserFragment; -use crate::context::InterAgentMessage; -use crate::context::InterAgentMessageType; +use crate::agent::control::AgentMessage; use crate::function_tool::FunctionCallError; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; @@ -22,7 +20,6 @@ use codex_protocol::items::SubAgentActivityItem; use codex_protocol::items::TurnItem; use codex_protocol::models::ResponseInputItem; use codex_protocol::openai_models::ReasoningEffort; -use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::SubAgentActivityKind; use codex_tools::ToolName; use serde::Deserialize; @@ -55,31 +52,16 @@ pub(crate) async fn emit_sub_agent_activity( session.emit_turn_item_completed(turn, item).await; } -fn communication_from_tool_message( - author: AgentPath, - recipient: AgentPath, +fn agent_message_from_tool( message: String, source: &crate::tools::context::ToolCallSource, - trigger_turn: bool, -) -> InterAgentCommunication { - if !matches!( +) -> AgentMessage { + if matches!( source, crate::tools::context::ToolCallSource::DirectPlaintextMessage ) { - return InterAgentCommunication::new_encrypted( - author, - recipient, - Vec::new(), - message, - trigger_turn, - ); - } - let message_type = if trigger_turn { - InterAgentMessageType::NewTask + AgentMessage::Plaintext(message) } else { - InterAgentMessageType::Message - }; - let content = - InterAgentMessage::new(message_type, recipient.clone(), author.clone(), message).render(); - InterAgentCommunication::new(author, recipient, Vec::new(), content, trigger_turn) + AgentMessage::Encrypted(message) + } } diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs index 19de4d842f..161e644d7e 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/followup_task.rs @@ -1,8 +1,8 @@ use super::analytics::ToolCallAnalytics; use super::message_tool::FollowupTaskArgs; -use super::message_tool::MessageDeliveryMode; use super::message_tool::handle_message_string_tool; use super::*; +use crate::agent::control::MessageDeliveryMode; use crate::tools::handlers::multi_agents_spec::create_followup_task_tool; use codex_tools::ToolSpec; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs index 1c7e9c1e99..e319a20183 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/message_tool.rs @@ -5,26 +5,10 @@ use super::analytics::ToolCallAnalytics; use super::*; -use crate::agent::child_config::build_agent_resume_config; -use crate::agent_communication::AgentCommunicationContext; -use crate::agent_communication::AgentCommunicationKind; +use crate::agent::control::MessageDeliveryError; +use crate::agent::control::MessageDeliveryMode; use crate::tools::context::FunctionToolOutput; -#[derive(Clone, Copy, PartialEq, Eq)] -pub(crate) enum MessageDeliveryMode { - QueueOnly, - TriggerTurn, -} - -impl MessageDeliveryMode { - fn trigger_turn(self) -> bool { - match self { - Self::QueueOnly => false, - Self::TriggerTurn => true, - } - } -} - #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] /// Input for the MultiAgentV2 `send_message` tool. @@ -68,68 +52,23 @@ pub(super) async fn handle_message_string_tool( } = invocation; let receiver_thread_id = resolve_agent_target(&session, &turn, &target).await?; analytics.set_receiver(receiver_thread_id); - let receiver_agent = session + let receiver_agent_path = session .services .agent_control - .ensure_agent_known(receiver_thread_id) - .map_err(|err| collab_agent_error(receiver_thread_id, err))?; - if mode == MessageDeliveryMode::TriggerTurn - && receiver_agent - .agent_path - .as_ref() - .is_some_and(AgentPath::is_root) - { - return Err(FunctionCallError::RespondToModel( - "Follow-up tasks can't target the root agent".to_string(), - )); - } - let receiver_agent_path = receiver_agent.agent_path.clone().ok_or_else(|| { - FunctionCallError::RespondToModel("target agent is missing an agent_path".to_string()) - })?; - let resume_config = - build_agent_resume_config(turn.as_ref()).map_err(FunctionCallError::RespondToModel)?; - session - .services - .agent_control - .ensure_v2_agent_loaded(resume_config, receiver_thread_id, /*parent*/ None) - .await - .map_err(|err| collab_agent_error(receiver_thread_id, err))?; - let author = turn - .session_source - .get_agent_path() - .unwrap_or_else(AgentPath::root); - let communication = communication_from_tool_message( - author, - receiver_agent_path.clone(), - message, - &source, - mode.trigger_turn(), - ); - let kind = match mode { - MessageDeliveryMode::QueueOnly => AgentCommunicationKind::Message, - MessageDeliveryMode::TriggerTurn => AgentCommunicationKind::Followup, - }; - let context = AgentCommunicationContext::new(kind, session.thread_id); - let parent_turn_id = - matches!(mode, MessageDeliveryMode::TriggerTurn).then(|| turn.sub_id.clone()); - let result = session - .services - .agent_control - .send_inter_agent_communication( + .deliver_message( + session.thread_id, + &turn, receiver_thread_id, - communication, - context, - crate::TurnStartOptions { - parent_turn_id, - root_turn_id: turn.turn_metadata_state.root_turn_id(), - turn_trigger: turn.turn_metadata_state.current_turn_trigger(), - cyber_access_program: turn.cyber_access_program, - ..Default::default() - }, + agent_message_from_tool(message, &source), + mode, ) .await - .map_err(|err| collab_agent_error(receiver_thread_id, err)); - result?; + .map_err(|err| match err { + MessageDeliveryError::InvalidRequest(message) => { + FunctionCallError::RespondToModel(message) + } + MessageDeliveryError::Agent(err) => collab_agent_error(receiver_thread_id, err), + })?; emit_sub_agent_activity( &session, &turn, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs index 70fc7d4350..f1987170eb 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/send_message.rs @@ -1,8 +1,8 @@ use super::analytics::ToolCallAnalytics; -use super::message_tool::MessageDeliveryMode; use super::message_tool::SendMessageArgs; use super::message_tool::handle_message_string_tool; use super::*; +use crate::agent::control::MessageDeliveryMode; use crate::tools::handlers::multi_agents_spec::create_send_message_tool; use codex_tools::ToolSpec; diff --git a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs index 9baf45b8ff..d7899c7a81 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_v2/spawn.rs @@ -2,6 +2,7 @@ use super::*; use crate::agent::child_config::SpawnConfigOptions; use crate::agent::child_config::SpawnConfigVersion; use crate::agent::child_config::prepare_agent_spawn_config; +use crate::agent::control::MessageDeliveryMode; use crate::agent::control::SpawnAgentForkMode; use crate::agent::control::SpawnAgentOptions; use crate::agent::next_thread_spawn_depth; @@ -156,12 +157,10 @@ async fn handle_spawn_agent( .session_source .get_agent_path() .unwrap_or_else(AgentPath::root); - let communication = communication_from_tool_message( + let communication = agent_message_from_tool(message, &source).into_communication( author, new_agent_path.clone(), - message, - &source, - /*trigger_turn*/ true, + MessageDeliveryMode::TriggerTurn, ); let context = AgentCommunicationContext::new(AgentCommunicationKind::Spawn, session.thread_id); let multi_agent_v2_usage_hints =