diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 49bbab0f87..1f96175d1f 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -183,12 +183,28 @@ impl AgentControl { agent_id: ThreadId, communication: InterAgentCommunication, ) -> CodexResult { - let last_task_message = last_task_message_from_communication(&communication); let state = self.upgrade()?; - let op = Op::InterAgentCommunication { communication }; - self.ensure_execution_capacity_for_op(agent_id, &op).await?; + self.ensure_execution_capacity_for_communication(agent_id, &communication) + .await?; + self.submit_inter_agent_communication(agent_id, &state, communication) + .await + } + + async fn submit_inter_agent_communication( + &self, + agent_id: ThreadId, + state: &Arc, + communication: InterAgentCommunication, + ) -> CodexResult { + let last_task_message = last_task_message_from_communication(&communication); let result = self - .handle_thread_request_result(agent_id, &state, state.send_op(agent_id, op).await) + .handle_thread_request_result( + agent_id, + state, + state + .send_op(agent_id, Op::InterAgentCommunication { communication }) + .await, + ) .await; if result.is_ok() { match last_task_message { diff --git a/codex-rs/core/src/agent/control/execution.rs b/codex-rs/core/src/agent/control/execution.rs index 42aaa50e02..28783e9304 100644 --- a/codex-rs/core/src/agent/control/execution.rs +++ b/codex-rs/core/src/agent/control/execution.rs @@ -2,6 +2,7 @@ use super::AgentControl; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; +use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; @@ -32,9 +33,28 @@ impl AgentControl { thread_id: ThreadId, op: &Op, ) -> CodexResult<()> { - if !op_starts_turn(op) { + match op { + Op::InterAgentCommunication { communication } => { + self.ensure_execution_capacity_for_communication(thread_id, communication) + .await + } + Op::UserInput { .. } => self.ensure_execution_capacity_for_thread(thread_id).await, + _ => Ok(()), + } + } + + pub(super) async fn ensure_execution_capacity_for_communication( + &self, + thread_id: ThreadId, + communication: &InterAgentCommunication, + ) -> CodexResult<()> { + if !communication.trigger_turn { return Ok(()); } + self.ensure_execution_capacity_for_thread(thread_id).await + } + + async fn ensure_execution_capacity_for_thread(&self, thread_id: ThreadId) -> CodexResult<()> { let state = self.upgrade()?; let thread = state.get_thread(thread_id).await?; if thread.codex.session.active_turn.lock().await.is_some() { @@ -92,11 +112,6 @@ impl AgentExecutionLimiter { } } -fn op_starts_turn(op: &Op) -> bool { - matches!(op, Op::UserInput { .. }) - || matches!(op, Op::InterAgentCommunication { communication } if communication.trigger_turn) -} - fn is_execution_limited( multi_agent_version: MultiAgentVersion, session_source: &SessionSource, diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index c4c40577ca..168b74ed24 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -356,8 +356,20 @@ impl AgentControl { ) .await; - self.send_input_after_capacity_check(new_thread.thread_id, &state, initial_operation) - .await?; + match (multi_agent_version, initial_operation) { + (MultiAgentVersion::V2, Op::InterAgentCommunication { communication }) => { + self.submit_inter_agent_communication(new_thread.thread_id, &state, communication) + .await?; + } + (_, initial_operation) => { + self.send_input_after_capacity_check( + new_thread.thread_id, + &state, + initial_operation, + ) + .await?; + } + } if multi_agent_version != MultiAgentVersion::V2 { let child_reference = agent_metadata .agent_path