fix: reject replay when stored tasks differ

lib/crewai/src/crewai/crew.py

@@ -2031,6 +2031,46 @@ def _find_task_index(task_id: str, stored_outputs: list[Any]) -> int | None:
             None,
         )
 
+    def _validate_replay_tasks(
+        self, stored_outputs: list[Any], start_index: int
+    ) -> None:
+        """Ensure stored outputs still correspond to the tasks that will receive them."""
+        if len(self.tasks) <= start_index:
+            raise ValueError(
+                "Cannot replay because the current crew does not match the stored task outputs."
+            )
+
+        stored_prefix = stored_outputs[: start_index + 1]
+        stored_identities = [
+            (
+                stored_output["output"].get("description"),
+                stored_output.get("expected_output"),
+            )
+            for stored_output in stored_prefix
+        ]
+        current_identities = [
+            (task.description, task.expected_output)
+            for task in self.tasks[: start_index + 1]
+        ]
+        if len(set(stored_identities)) != len(stored_identities) or len(
+            set(current_identities)
+        ) != len(current_identities):
+            raise ValueError(
+                "Cannot replay because the stored task identities are ambiguous."
+            )
+
+        for index, stored_output in enumerate(stored_prefix):
+            task = self.tasks[index]
+            output = stored_output["output"]
+            stored_expected_output = stored_output.get("expected_output")
+            if task.description != output.get("description") or (
+                stored_expected_output is not None
+                and task.expected_output != stored_expected_output
+            ):
+                raise ValueError(
+                    "Cannot replay because the current crew does not match the stored task outputs."
+                )
+
     def replay(self, task_id: str, inputs: dict[str, Any] | None = None) -> CrewOutput:
         """Replay the crew execution from a specific task."""
         stored_outputs = self._task_output_handler.load()
@@ -2042,6 +2082,8 @@ def replay(self, task_id: str, inputs: dict[str, Any] | None = None) -> CrewOutp
         if start_index is None:
             raise ValueError(f"Task with id {task_id} not found in the crew's tasks.")
 
+        self._validate_replay_tasks(stored_outputs, start_index)
+
         replay_inputs = (
             inputs if inputs is not None else stored_outputs[start_index]["inputs"]
         )

lib/crewai/tests/test_crew.py

@@ -3045,23 +3045,158 @@ def test_replay_feature(researcher, writer):
     )
 
     with patch.object(Task, "execute_sync") as mock_execute_task:
-        mock_execute_task.return_value = TaskOutput(
-            description="Mock description",
-            raw="Mocked output for list of ideas",
-            agent="Researcher",
-            json_dict=None,
-            output_format=OutputFormat.RAW,
-            pydantic=None,
-            summary="Mocked output for list of ideas",
-            messages=[],
-        )
+        mock_execute_task.side_effect = [
+            TaskOutput(
+                description=list_ideas.description,
+                raw="Mocked output for list of ideas",
+                agent="Researcher",
+                json_dict=None,
+                output_format=OutputFormat.RAW,
+                pydantic=None,
+                summary="Mocked output for list of ideas",
+                messages=[],
+            ),
+            TaskOutput(
+                description=write.description,
+                raw="Mocked output for list of ideas",
+                agent="Researcher",
+                json_dict=None,
+                output_format=OutputFormat.RAW,
+                pydantic=None,
+                summary="Mocked output for list of ideas",
+                messages=[],
+            ),
+            TaskOutput(
+                description=write.description,
+                raw="Mocked output for list of ideas",
+                agent="Researcher",
+                json_dict=None,
+                output_format=OutputFormat.RAW,
+                pydantic=None,
+                summary="Mocked output for list of ideas",
+                messages=[],
+            ),
+        ]
 
         crew.kickoff()
         crew.replay(str(write.id))
         # Ensure context was passed correctly
         assert mock_execute_task.call_count == 3
 
 
+def test_replay_rejects_changed_task_order(researcher):
+    """Replay must not restore a saved output onto a different current task."""
+    research = Task(
+        description="Research the topic",
+        expected_output="Research notes",
+        agent=researcher,
+    )
+    write = Task(
+        description="Write the article",
+        expected_output="An article",
+        agent=researcher,
+    )
+    plan = Task(
+        description="Plan the article",
+        expected_output="An outline",
+        agent=researcher,
+    )
+    crew = Crew(agents=[researcher], tasks=[plan, research, write])
+
+    stored_outputs = [
+        {
+            "task_id": str(research.id),
+            "expected_output": research.expected_output,
+            "output": {"description": research.description},
+            "inputs": {},
+        },
+        {
+            "task_id": str(write.id),
+            "expected_output": write.expected_output,
+            "output": {"description": write.description},
+            "inputs": {},
+        },
+    ]
+    with patch(
+        "crewai.utilities.task_output_storage_handler.TaskOutputStorageHandler.load",
+        return_value=stored_outputs,
+    ):
+        with pytest.raises(ValueError, match="current crew does not match"):
+            crew.replay(str(write.id))
+
+
+def test_replay_rejects_reordered_tasks_with_matching_expected_output(researcher):
+    """Task descriptions keep replay from confusing tasks with the same expected output."""
+    research = Task(
+        description="Research the topic",
+        expected_output="A report",
+        agent=researcher,
+    )
+    write = Task(
+        description="Write the article",
+        expected_output="A report",
+        agent=researcher,
+    )
+    crew = Crew(agents=[researcher], tasks=[write, research])
+
+    stored_outputs = [
+        {
+            "task_id": str(research.id),
+            "expected_output": research.expected_output,
+            "output": {"description": research.description},
+            "inputs": {},
+        },
+        {
+            "task_id": str(write.id),
+            "expected_output": write.expected_output,
+            "output": {"description": write.description},
+            "inputs": {},
+        },
+    ]
+    with patch(
+        "crewai.utilities.task_output_storage_handler.TaskOutputStorageHandler.load",
+        return_value=stored_outputs,
+    ):
+        with pytest.raises(ValueError, match="current crew does not match"):
+            crew.replay(str(write.id))
+
+
+def test_replay_rejects_ambiguous_task_identities(researcher):
+    """Replay must fail loud when persisted task details cannot identify a task."""
+    first_task = Task(
+        description="Write a report",
+        expected_output="A report",
+        agent=researcher,
+    )
+    second_task = Task(
+        description="Write a report",
+        expected_output="A report",
+        agent=researcher,
+    )
+    crew = Crew(agents=[researcher], tasks=[second_task, first_task])
+
+    stored_outputs = [
+        {
+            "task_id": str(first_task.id),
+            "expected_output": first_task.expected_output,
+            "output": {"description": first_task.description},
+            "inputs": {},
+        },
+        {
+            "task_id": str(second_task.id),
+            "expected_output": second_task.expected_output,
+            "output": {"description": second_task.description},
+            "inputs": {},
+        },
+    ]
+    with patch(
+        "crewai.utilities.task_output_storage_handler.TaskOutputStorageHandler.load",
+        return_value=stored_outputs,
+    ):
+        with pytest.raises(ValueError, match="task identities are ambiguous"):
+            crew.replay(str(second_task.id))
+
+
 @pytest.mark.vcr()
 def test_crew_replay_error(researcher, writer):
     task = Task(
@@ -3292,7 +3427,7 @@ def test_replay_with_context():
     )
 
     context_output = TaskOutput(
-        description="Context Task Output",
+        description=task1.description,
         agent="test_agent",
         raw="context raw output",
         pydantic=None,
@@ -3309,6 +3444,7 @@ def test_replay_with_context():
         return_value=[
             {
                 "task_id": str(task1.id),
+                "expected_output": task1.expected_output,
                 "output": {
                     "description": context_output.description,
                     "summary": context_output.summary,
@@ -3322,8 +3458,9 @@ def test_replay_with_context():
             },
             {
                 "task_id": str(task2.id),
+                "expected_output": task2.expected_output,
                 "output": {
-                    "description": "Test Task Output",
+                    "description": task2.description,
                     "summary": None,
                     "raw": "test raw output",
                     "pydantic": None,