fix: handle duplicate task ordering

lib/crewai/src/crewai/crew.py

@@ -955,10 +955,16 @@ class Crew(FlowTrackable, BaseModel):
             tasks=self.tasks, planning_agent_llm=self.planning_llm
         )._handle_crew_planning()
 
-        plan_map: dict[int, str] = {
-            step_plan.task_number: step_plan.plan
-            for step_plan in result.list_of_plans_per_task
-        }
+        plan_map: dict[int, str] = {}
+        for step_plan in result.list_of_plans_per_task:
+            if step_plan.task_number in plan_map:
+                self._logger.log(
+                    "warning",
+                    f"Duplicate plan for Task Number {step_plan.task_number}, "
+                    "using the first plan",
+                )
+            else:
+                plan_map[step_plan.task_number] = step_plan.plan
 
         for idx, task in enumerate(self.tasks):
             task_number = idx + 1

lib/crewai/tests/utilities/test_planning_handler.py

@@ -326,3 +326,38 @@ class TestCrewPlanningIntegration:
 
         assert "Original task description" in task.description
         assert "Additional plan steps" in task.description
+
+    def test_crew_planning_with_duplicate_task_numbers(self):
+        """Test that duplicate task numbers use the first plan and log a warning."""
+        agent = Agent(role="Agent 1", goal="Goal 1", backstory="Backstory 1")
+
+        task = Task(
+            description="Task description",
+            expected_output="Output 1",
+            agent=agent,
+        )
+
+        crew = Crew(
+            agents=[agent],
+            tasks=[task],
+            planning=True,
+        )
+
+        # Two plans with the same task_number - should use the first one
+        duplicate_plans = [
+            PlanPerTask(task_number=1, task="Task 1", plan=" [FIRST PLAN]"),
+            PlanPerTask(task_number=1, task="Task 1", plan=" [SECOND PLAN]"),
+        ]
+
+        mock_planner_result = PlannerTaskPydanticOutput(
+            list_of_plans_per_task=duplicate_plans
+        )
+
+        with patch.object(
+            CrewPlanner, "_handle_crew_planning", return_value=mock_planner_result
+        ):
+            crew._handle_crew_planning()
+
+        # Should use the first plan, not the second
+        assert "[FIRST PLAN]" in task.description
+        assert "[SECOND PLAN]" not in task.description