Fix task.copy() to preserve NOT_SPECIFIED context (fixes #3691)

- 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>
This commit is contained in:
Devin AI
2025-10-10 18:27:17 +00:00
parent 29919c2d81
commit fe86049bd7
2 changed files with 63 additions and 7 deletions

View File

@@ -7,7 +7,7 @@ import uuid
import warnings import warnings
from collections.abc import Callable from collections.abc import Callable
from concurrent.futures import Future from concurrent.futures import Future
from copy import copy from copy import copy as shallow_copy
from hashlib import md5 from hashlib import md5
from pathlib import Path from pathlib import Path
from typing import ( from typing import (
@@ -671,17 +671,20 @@ Follow these guidelines:
copied_data = self.model_dump(exclude=exclude) copied_data = self.model_dump(exclude=exclude)
copied_data = {k: v for k, v in copied_data.items() if v is not None} copied_data = {k: v for k, v in copied_data.items() if v is not None}
cloned_context = ( if self.context is NOT_SPECIFIED:
[task_mapping[context_task.key] for context_task in self.context] cloned_context = self.context
if isinstance(self.context, list) else:
else None 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]: def get_agent_by_role(role: str) -> Union["BaseAgent", None]:
return next((agent for agent in agents if agent.role == role), 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_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__( return self.__class__(
**copied_data, **copied_data,

View File

@@ -900,6 +900,59 @@ def test_conditional_task_copy_preserves_type():
assert isinstance(copied_conditional_task, ConditionalTask) 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): def test_interpolate_inputs(tmp_path):
task = Task( task = Task(
description="Give me a list of 5 interesting ideas about {topic} to explore for an article, what makes them unique and interesting.", description="Give me a list of 5 interesting ideas about {topic} to explore for an article, what makes them unique and interesting.",