fix: detect circular task context dependencies

lib/crewai/src/crewai/crew.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Callable, Sequence
+from collections.abc import Callable, Iterator, Sequence
 from concurrent.futures import Future
 from copy import copy as shallow_copy
 from hashlib import md5
@@ -835,6 +835,55 @@ def validate_async_task_cannot_include_sequential_async_tasks_in_context(
                                 break
         return self
 
+    @model_validator(mode="after")
+    def validate_context_no_circular_dependencies(self) -> Self:
+        """Validates that task context dependencies do not contain cycles."""
+        task_ids = {id(task) for task in self.tasks}
+        visited: set[int] = set()
+
+        def iter_context_tasks(task: Task) -> Iterator[Task]:
+            if isinstance(task.context, list):
+                for context_task in task.context:
+                    if id(context_task) in task_ids:
+                        yield context_task
+
+        for task in self.tasks:
+            if id(task) in visited:
+                continue
+
+            path: list[Task] = [task]
+            on_path: dict[int, int] = {id(task): 0}
+            stack: list[tuple[Task, Iterator[Task]]] = [
+                (task, iter_context_tasks(task))
+            ]
+
+            while stack:
+                current_task, context_tasks = stack[-1]
+                for context_task in context_tasks:
+                    context_task_id = id(context_task)
+                    if context_task_id in on_path:
+                        cycle_tasks = [*path[on_path[context_task_id] :], context_task]
+                        cycle_descriptions = " -> ".join(
+                            cycle_task.description for cycle_task in cycle_tasks
+                        )
+                        raise ValueError(
+                            "Task context dependencies contain a circular dependency: "
+                            f"{cycle_descriptions}."
+                        )
+                    if context_task_id in visited:
+                        continue
+
+                    on_path[context_task_id] = len(path)
+                    path.append(context_task)
+                    stack.append((context_task, iter_context_tasks(context_task)))
+                    break
+                else:
+                    stack.pop()
+                    path.pop()
+                    on_path.pop(id(current_task), None)
+                    visited.add(id(current_task))
+        return self
+
     @model_validator(mode="after")
     def validate_context_no_future_tasks(self) -> Self:
         """Validates that a task's context does not include future tasks."""

lib/crewai/tests/test_crew.py

@@ -253,6 +253,58 @@ def test_context_no_future_tasks(researcher, writer):
         Crew(tasks=[task1, task2, task3, task4], agents=[researcher, writer])
 
 
+def test_context_no_circular_dependencies(researcher, writer):
+    task1 = Task(
+        description="Task 1",
+        expected_output="output",
+        agent=researcher,
+    )
+    task2 = Task(
+        description="Task 2",
+        expected_output="output",
+        agent=researcher,
+    )
+    task3 = Task(
+        description="Task 3",
+        expected_output="output",
+        agent=researcher,
+    )
+
+    task1.context = [task2]
+    task2.context = [task3]
+    task3.context = [task1]
+
+    with pytest.raises(
+        ValueError,
+        match=re.escape(
+            "Task context dependencies contain a circular dependency: "
+            "Task 1 -> Task 2 -> Task 3 -> Task 1."
+        ),
+    ):
+        Crew(tasks=[task1, task2, task3], agents=[researcher, writer])
+
+
+def test_context_no_circular_dependencies_handles_long_chains(researcher, writer):
+    tasks = [
+        Task(
+            description=f"Task {index}",
+            expected_output="output",
+            agent=researcher,
+        )
+        for index in range(1100)
+    ]
+    for task, context_task in zip(tasks, tasks[1:]):
+        task.context = [context_task]
+
+    with pytest.raises(
+        ValueError,
+        match=re.escape(
+            "Task 'Task 0' has a context dependency on a future task 'Task 1', which is not allowed."
+        ),
+    ):
+        Crew(tasks=tasks, agents=[researcher, writer])
+
+
 def test_crew_config_with_wrong_keys():
     no_tasks_config = json.dumps(
         {