Move V2 agent message delivery into AgentControl (#45670)

## What changed

Extract target validation, runtime reloading, and message delivery from the V2 tool handler into `AgentControl::deliver_message`. Keep target resolution, analytics, and tool-facing error mapping in the handler.

Represent plaintext and encrypted payloads with `AgentMessage`, sharing communication construction with agent spawning. Preserve queue-only and follow-up turn semantics, target checks before reload, and turn metadata propagation.

GitOrigin-RevId: a88bcede015a17f6c11c7cad95ee58add5dc2820

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;

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<AgentPath, MessageDeliveryError> {
+        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)
+    }
+}

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)
+    }
 }

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;
 

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,

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;
 

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 =