From fe86049bd7ae03f40ea2aeb8ca325738096f8df2 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Fri, 10 Oct 2025 18:27:17 +0000
Subject: [PATCH] Fix task.copy() to preserve NOT_SPECIFIED context (fixes
 #3691)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Add check in task.copy() to preserve NOT_SPECIFIED context value
- Previously, NOT_SPECIFIED was being converted to None during copy
- This affected kickoff_for_each and kickoff_for_each_async methods
- Add comprehensive tests covering NOT_SPECIFIED, list, and None contexts
- Rename import 'copy' to 'shallow_copy' for clarity

Co-Authored-By: João <joao@crewai.com>
---
 src/crewai/task.py | 17 +++++++++------
 tests/test_task.py | 53 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 63 insertions(+), 7 deletions(-)

diff --git a/src/crewai/task.py b/src/crewai/task.py
index ebf284317..7efaa85eb 100644
--- a/src/crewai/task.py
+++ b/src/crewai/task.py
@@ -7,7 +7,7 @@ import uuid
 import warnings
 from collections.abc import Callable
 from concurrent.futures import Future
-from copy import copy
+from copy import copy as shallow_copy
 from hashlib import md5
 from pathlib import Path
 from typing import (
@@ -671,17 +671,20 @@ Follow these guidelines:
         copied_data = self.model_dump(exclude=exclude)
         copied_data = {k: v for k, v in copied_data.items() if v is not None}
 
-        cloned_context = (
-            [task_mapping[context_task.key] for context_task in self.context]
-            if isinstance(self.context, list)
-            else None
-        )
+        if self.context is NOT_SPECIFIED:
+            cloned_context = self.context
+        else:
+            cloned_context = (
+                [task_mapping[context_task.key] for context_task in self.context]
+                if isinstance(self.context, list)
+                else None
+            )
 
         def get_agent_by_role(role: str) -> Union["BaseAgent", None]:
             return next((agent for agent in agents if agent.role == role), None)
 
         cloned_agent = get_agent_by_role(self.agent.role) if self.agent else None
-        cloned_tools = copy(self.tools) if self.tools else []
+        cloned_tools = shallow_copy(self.tools) if self.tools else []
 
         return self.__class__(
             **copied_data,
diff --git a/tests/test_task.py b/tests/test_task.py
index 0e304df54..83e4ed88b 100644
--- a/tests/test_task.py
+++ b/tests/test_task.py
@@ -900,6 +900,59 @@ def test_conditional_task_copy_preserves_type():
     assert isinstance(copied_conditional_task, ConditionalTask)
 
 
+def test_task_copy_preserves_not_specified_context():
+    """Test that copying a task preserves NOT_SPECIFIED context value."""
+    from crewai.utilities.constants import NOT_SPECIFIED
+
+    task = Task(
+        description="Test task",
+        expected_output="Test output"
+    )
+
+    assert task.context is NOT_SPECIFIED
+
+    copied_task = task.copy(agents=[], task_mapping={})
+
+    assert copied_task.context is NOT_SPECIFIED
+    assert copied_task.context is not None
+
+
+def test_task_copy_with_list_context():
+    """Test that copying a task with list context works correctly."""
+    task1 = Task(
+        description="Task 1",
+        expected_output="Output 1"
+    )
+    task2 = Task(
+        description="Task 2",
+        expected_output="Output 2",
+        context=[task1]
+    )
+
+    task_mapping = {task1.key: task1}
+
+    copied_task2 = task2.copy(agents=[], task_mapping=task_mapping)
+
+    assert isinstance(copied_task2.context, list)
+    assert len(copied_task2.context) == 1
+    assert copied_task2.context[0] is task1
+
+
+def test_task_copy_with_none_context():
+    """Test that copying a task with None context works correctly."""
+    task = Task(
+        description="Test task",
+        expected_output="Test output",
+        context=None
+    )
+
+    assert task.context is None
+
+    copied_task = task.copy(agents=[], task_mapping={})
+
+    assert copied_task.context is None
+
+
 def test_interpolate_inputs(tmp_path):
     task = Task(
         description="Give me a list of 5 interesting ideas about {topic} to explore for an article, what makes them unique and interesting.",