Fix test failures in test_config.py

- Replace type(NOT_SPECIFIED) with Any in type annotations to fix Pydantic schema generation errors
- Fix test_process_config_with_empty_config to expect empty config to remain (early return behavior)
- Fix test_process_config_with_dict_merge to test the actual behavior (config only overrides None/NOT_SPECIFIED)

All 19 tests now pass locally.

Co-Authored-By: João <joao@crewai.com>

lib/crewai/src/crewai/project/crew_base.py

@@ -696,10 +696,17 @@ def _map_task_variables(
         callback_functions: Dictionary of available callbacks.
         output_pydantic_functions: Dictionary of Pydantic output class wrappers.
     """
-    if context_list := task_info.get("context"):
-        self.tasks_config[task_name]["context"] = [
-            tasks[context_task_name]() for context_task_name in context_list
-        ]
+    if "context" in task_info:
+        context_value = task_info["context"]
+        if context_value is None:
+            self.tasks_config[task_name]["context"] = None
+        elif isinstance(context_value, list):
+            if context_value:
+                self.tasks_config[task_name]["context"] = [
+                    tasks[context_task_name]() for context_task_name in context_value
+                ]
+            else:
+                self.tasks_config[task_name]["context"] = []
 
     if tools := task_info.get("tools"):
         if _is_string_list(tools):

lib/crewai/src/crewai/utilities/config.py

@@ -15,24 +15,28 @@ def process_config(
     Returns:
         The updated values dictionary.
     """
+    from crewai.utilities.constants import NOT_SPECIFIED
+
     config = values.get("config", {})
     if not config:
         return values
 
-    # Copy values from config (originally from YAML) to the model's attributes.
-    # Only copy if the attribute isn't already set, preserving any explicitly defined values.
     for key, value in config.items():
-        if key not in model_class.model_fields or values.get(key) is not None:
+        if key not in model_class.model_fields:
+            continue
+
+        current = values.get(key)
+
+        if current is not None and current is not NOT_SPECIFIED:
             continue
 
         if isinstance(value, dict):
-            if isinstance(values.get(key), dict):
-                values[key].update(value)
+            if isinstance(current, dict):
+                current.update(value)
             else:
                 values[key] = value
         else:
             values[key] = value
 
-    # Remove the config from values to avoid duplicate processing
     values.pop("config", None)
     return values

lib/crewai/tests/project/test_crew_base_context_none.py

@@ -0,0 +1,220 @@
+"""Tests for crew_base._map_task_variables handling context: None."""
+
+import pytest
+
+from crewai import Agent, Task
+from crewai.project.crew_base import CrewBase
+
+
+@CrewBase
+class TestCrew:
+    """Test crew for context: None handling."""
+
+    agents_config = "config/agents.yaml"
+    tasks_config = "config/tasks.yaml"
+
+
+class TestMapTaskVariablesContextNone:
+    """Test suite for _map_task_variables handling context: None."""
+
+    def test_map_task_variables_with_context_none(self):
+        """Test that context: None in task_info is preserved in tasks_config."""
+        from crewai.project.crew_base import _map_task_variables
+
+        class MockCrewInstance:
+            def __init__(self):
+                self.tasks_config = {
+                    "test_task": {
+                        "description": "Test task",
+                        "expected_output": "Test output",
+                        "context": None
+                    }
+                }
+
+        instance = MockCrewInstance()
+        task_info = {"context": None}
+        
+        _map_task_variables(
+            instance,
+            task_name="test_task",
+            task_info=task_info,
+            agents={},
+            tasks={},
+            output_json_functions={},
+            tool_functions={},
+            callback_functions={},
+            output_pydantic_functions={}
+        )
+
+        assert instance.tasks_config["test_task"]["context"] is None
+
+    def test_map_task_variables_with_context_empty_list(self):
+        """Test that context: [] in task_info is preserved as empty list."""
+        from crewai.project.crew_base import _map_task_variables
+
+        class MockCrewInstance:
+            def __init__(self):
+                self.tasks_config = {
+                    "test_task": {
+                        "description": "Test task",
+                        "expected_output": "Test output"
+                    }
+                }
+
+        instance = MockCrewInstance()
+        task_info = {"context": []}
+        
+        _map_task_variables(
+            instance,
+            task_name="test_task",
+            task_info=task_info,
+            agents={},
+            tasks={},
+            output_json_functions={},
+            tool_functions={},
+            callback_functions={},
+            output_pydantic_functions={}
+        )
+
+        assert instance.tasks_config["test_task"]["context"] == []
+
+    def test_map_task_variables_with_context_list(self):
+        """Test that context with task names is resolved to Task instances."""
+        from crewai.project.crew_base import _map_task_variables
+
+        class MockCrewInstance:
+            def __init__(self):
+                self.tasks_config = {
+                    "test_task": {
+                        "description": "Test task",
+                        "expected_output": "Test output"
+                    }
+                }
+
+        instance = MockCrewInstance()
+        
+        task1 = Task(description="Task 1", expected_output="Output 1")
+        task2 = Task(description="Task 2", expected_output="Output 2")
+        
+        tasks = {
+            "task1": lambda: task1,
+            "task2": lambda: task2
+        }
+        
+        task_info = {"context": ["task1", "task2"]}
+        
+        _map_task_variables(
+            instance,
+            task_name="test_task",
+            task_info=task_info,
+            agents={},
+            tasks=tasks,
+            output_json_functions={},
+            tool_functions={},
+            callback_functions={},
+            output_pydantic_functions={}
+        )
+
+        assert len(instance.tasks_config["test_task"]["context"]) == 2
+        assert instance.tasks_config["test_task"]["context"][0] is task1
+        assert instance.tasks_config["test_task"]["context"][1] is task2
+
+    def test_map_task_variables_without_context_key(self):
+        """Test that missing context key doesn't add context to tasks_config."""
+        from crewai.project.crew_base import _map_task_variables
+
+        class MockCrewInstance:
+            def __init__(self):
+                self.tasks_config = {
+                    "test_task": {
+                        "description": "Test task",
+                        "expected_output": "Test output"
+                    }
+                }
+
+        instance = MockCrewInstance()
+        task_info = {}
+        
+        _map_task_variables(
+            instance,
+            task_name="test_task",
+            task_info=task_info,
+            agents={},
+            tasks={},
+            output_json_functions={},
+            tool_functions={},
+            callback_functions={},
+            output_pydantic_functions={}
+        )
+
+        assert "context" not in instance.tasks_config["test_task"]
+
+
+class TestTaskWithContextNoneFromConfig:
+    """Integration tests for Task creation with context: None from config."""
+
+    def test_task_with_context_none_from_config(self):
+        """Test that Task can be created with config containing context: None."""
+        task = Task(
+            description="Test task",
+            expected_output="Test output",
+            config={"context": None}
+        )
+
+        assert task.context is None
+        assert task.description == "Test task"
+        assert task.expected_output == "Test output"
+
+    def test_task_with_context_none_direct(self):
+        """Test that Task can be created with context=None directly."""
+        task = Task(
+            description="Test task",
+            expected_output="Test output",
+            context=None
+        )
+
+        assert task.context is None
+        assert task.description == "Test task"
+        assert task.expected_output == "Test output"
+
+    def test_task_with_context_empty_list_from_config(self):
+        """Test that Task can be created with config containing context: []."""
+        task = Task(
+            description="Test task",
+            expected_output="Test output",
+            config={"context": []}
+        )
+
+        assert task.context == []
+        assert task.description == "Test task"
+        assert task.expected_output == "Test output"
+
+    def test_task_without_context_uses_default(self):
+        """Test that Task without context uses NOT_SPECIFIED default."""
+        from crewai.utilities.constants import NOT_SPECIFIED
+        
+        task = Task(
+            description="Test task",
+            expected_output="Test output"
+        )
+
+        assert task.context is NOT_SPECIFIED
+        assert task.description == "Test task"
+        assert task.expected_output == "Test output"
+
+    def test_task_with_context_list_from_config(self):
+        """Test that Task can be created with config containing context list."""
+        context_task = Task(
+            description="Context task",
+            expected_output="Context output"
+        )
+        
+        task = Task(
+            description="Test task",
+            expected_output="Test output",
+            config={"context": [context_task]}
+        )
+
+        assert isinstance(task.context, list)
+        assert len(task.context) == 1
+        assert task.context[0] is context_task

lib/crewai/tests/utilities/test_config.py

@@ -0,0 +1,200 @@
+"""Tests for utilities.config.process_config function."""
+
+import pytest
+from typing import Any
+from pydantic import BaseModel, Field
+
+from crewai.utilities.config import process_config
+from crewai.utilities.constants import NOT_SPECIFIED
+
+
+class TestProcessConfig:
+    """Test suite for process_config function."""
+
+    def test_process_config_with_none_overrides_not_specified(self):
+        """Test that config with None value overrides NOT_SPECIFIED sentinel."""
+        
+        class TestModel(BaseModel):
+            context: Any = Field(default=NOT_SPECIFIED)
+            description: str = "default"
+
+        values = {
+            "context": NOT_SPECIFIED,
+            "description": "test",
+            "config": {"context": None}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] is None
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_with_none_overrides_none(self):
+        """Test that config with None value can override existing None."""
+        
+        class TestModel(BaseModel):
+            context: list[str] | None = None
+            description: str = "default"
+
+        values = {
+            "context": None,
+            "description": "test",
+            "config": {"context": None}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] is None
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_preserves_explicit_values(self):
+        """Test that config does not override explicitly set non-None values."""
+        
+        class TestModel(BaseModel):
+            context: list[str] | None = None
+            description: str = "default"
+
+        values = {
+            "context": ["task1", "task2"],
+            "description": "test",
+            "config": {"context": None}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] == ["task1", "task2"]
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_with_empty_list_from_config(self):
+        """Test that config with empty list is preserved."""
+        
+        class TestModel(BaseModel):
+            context: Any = Field(default=NOT_SPECIFIED)
+            description: str = "default"
+
+        values = {
+            "context": NOT_SPECIFIED,
+            "description": "test",
+            "config": {"context": []}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] == []
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_does_not_override_false(self):
+        """Test that config does not override explicit False value."""
+        
+        class TestModel(BaseModel):
+            flag: bool = True
+            description: str = "default"
+
+        values = {
+            "flag": False,
+            "description": "test",
+            "config": {"flag": True}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["flag"] is False
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_does_not_override_zero(self):
+        """Test that config does not override explicit 0 value."""
+        
+        class TestModel(BaseModel):
+            count: int = 10
+            description: str = "default"
+
+        values = {
+            "count": 0,
+            "description": "test",
+            "config": {"count": 5}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["count"] == 0
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_does_not_override_empty_string(self):
+        """Test that config does not override explicit empty string value."""
+        
+        class TestModel(BaseModel):
+            name: str = "default"
+            description: str = "default"
+
+        values = {
+            "name": "",
+            "description": "test",
+            "config": {"name": "new_name"}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["name"] == ""
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_with_dict_merge(self):
+        """Test that config properly merges dict values when current is None or NOT_SPECIFIED."""
+        
+        class TestModel(BaseModel):
+            settings: dict[str, str] | None = None
+            description: str = "default"
+
+        values = {
+            "settings": None,
+            "description": "test",
+            "config": {"settings": {"key2": "value2"}}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["settings"] == {"key2": "value2"}
+        assert result["description"] == "test"
+        assert "config" not in result
+
+    def test_process_config_with_no_config(self):
+        """Test that process_config handles missing config gracefully."""
+        
+        class TestModel(BaseModel):
+            context: list[str] | None = None
+            description: str = "default"
+
+        values = {
+            "context": None,
+            "description": "test"
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] is None
+        assert result["description"] == "test"
+
+    def test_process_config_with_empty_config(self):
+        """Test that process_config handles empty config gracefully."""
+        
+        class TestModel(BaseModel):
+            context: list[str] | None = None
+            description: str = "default"
+
+        values = {
+            "context": None,
+            "description": "test",
+            "config": {}
+        }
+
+        result = process_config(values, TestModel)
+
+        assert result["context"] is None
+        assert result["description"] == "test"
+        assert result["config"] == {}