fix(crew): identify tasks by object when copying instead of by key

lib/crewai/src/crewai/crew.py

@@ -2142,7 +2142,7 @@ def copy(self) -> Crew:  # type: ignore[override]
         manager_agent = self.manager_agent.copy() if self.manager_agent else None
         manager_llm = shallow_copy(self.manager_llm) if self.manager_llm else None
 
-        task_mapping: dict[str, Any] = {}
+        task_mapping: dict[int, Any] = {}
 
         cloned_tasks = []
         existing_knowledge_sources = shallow_copy(self.knowledge_sources)
@@ -2151,12 +2151,12 @@ def copy(self) -> Crew:  # type: ignore[override]
         for task in self.tasks:
             cloned_task = task.copy(cloned_agents, task_mapping)
             cloned_tasks.append(cloned_task)
-            task_mapping[task.key] = cloned_task
+            task_mapping[id(task)] = cloned_task
 
         for cloned_task, original_task in zip(cloned_tasks, self.tasks, strict=False):
             if isinstance(original_task.context, list):
                 cloned_context = [
-                    task_mapping[context_task.key]
+                    task_mapping.get(id(context_task), context_task)
                     for context_task in original_task.context
                 ]
                 cloned_task.context = cloned_context

lib/crewai/src/crewai/task.py

@@ -1152,7 +1152,7 @@ def increment_delegations(self, agent_name: str | None) -> None:
         self.delegations += 1
 
     def copy(  # type: ignore
-        self, agents: Sequence[BaseAgent], task_mapping: dict[str, Task]
+        self, agents: Sequence[BaseAgent], task_mapping: dict[int, Task]
     ) -> Task:
         """Creates a deep copy of the Task while preserving its original class type.
 
@@ -1176,7 +1176,10 @@ def copy(  # type: ignore
         cloned_context = (
             self.context
             if self.context is NOT_SPECIFIED
-            else [task_mapping[context_task.key] for context_task in self.context]
+            else [
+                task_mapping.get(id(context_task), context_task)
+                for context_task in self.context
+            ]
             if isinstance(self.context, list)
             else None
         )

lib/crewai/tests/test_crew.py

@@ -4989,3 +4989,40 @@ def test_memory_remember_receives_task_content():
     assert "Researcher" in raw
     assert "Expected result:" in raw
     assert "Result:" in raw
+
+
+def test_crew_copy_preserves_context_with_duplicate_task_text(researcher):
+    """Tasks sharing description/expected_output must not collide when copied.
+
+    ``Task.key`` is an md5 of description + expected_output, so identical tasks
+    share a key. Keying the clone mapping by that value made the second task
+    overwrite the first and re-wired context to the wrong clone.
+    """
+    first = Task(description="same", expected_output="same", agent=researcher)
+    second = Task(description="same", expected_output="same", agent=researcher)
+    third = Task(
+        description="third", expected_output="third", agent=researcher, context=[first]
+    )
+
+    assert first.key == second.key  # documents the collision this guards against
+
+    crew_copy = Crew(agents=[researcher], tasks=[first, second, third]).copy()
+
+    assert crew_copy.tasks[2].context[0] is crew_copy.tasks[0]
+
+
+def test_crew_copy_keeps_context_task_outside_the_crew(researcher):
+    """A context task that is not a crew member must not break copying.
+
+    ``validate_context_no_future_tasks`` skips context tasks that are not crew
+    members, so such a crew is valid and can be kicked off; copying it must not
+    raise.
+    """
+    outside = Task(description="outside", expected_output="o", agent=researcher)
+    inside = Task(
+        description="inside", expected_output="i", agent=researcher, context=[outside]
+    )
+
+    crew_copy = Crew(agents=[researcher], tasks=[inside]).copy()
+
+    assert crew_copy.tasks[0].context[0] is outside