diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 0f77b2d224..5c62e71ec8 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/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"] ) diff --git a/lib/crewai/tests/test_crew.py b/lib/crewai/tests/test_crew.py index 0195112cb9..0e9fed5860 100644 --- a/lib/crewai/tests/test_crew.py +++ b/lib/crewai/tests/test_crew.py @@ -3045,16 +3045,38 @@ 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)) @@ -3062,6 +3084,119 @@ def test_replay_feature(researcher, writer): 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,