Compare commits

...

2 Commits

Author SHA1 Message Date
Devin AI
ef7e565f71 Fix type-checker and lint CI failures
- Add explicit type hint for cloned_context to fix mypy error
- Fix pre-existing lint error: use raw string for regex pattern

Co-Authored-By: João <joao@crewai.com>
2025-10-10 18:33:39 +00:00
Devin AI
fe86049bd7 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>
2025-10-10 18:27:17 +00:00
2 changed files with 65 additions and 8 deletions

View File

@@ -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,21 @@ 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
)
cloned_context: list["Task"] | None | _NotSpecified
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,

View File

@@ -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.",
@@ -1218,7 +1271,7 @@ def test_create_directory_false():
assert not resolved_dir.exists()
with pytest.raises(
RuntimeError, match="Directory .* does not exist and create_directory is False"
RuntimeError, match=r"Directory .* does not exist and create_directory is False"
):
task._save_file("test content")