Add verbose per-step agent execution tracing (#5817)

- Add crew-level start/finish logs: [crew_name] Starting/Completed
- Add per-task start/completion logs: [Agent: X] Starting/Task complete
- Add handoff logs between agents: passing to / Received context from
- Support custom crew name in logs
- Use task.name when available, fallback to description
- Add completion/handoff logs for async tasks
- Add 7 unit tests covering all new verbose tracing behavior

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

lib/crewai/src/crewai/crew.py

@@ -952,6 +952,7 @@ class Crew(FlowTrackable, BaseModel):
 
         try:
             inputs = prepare_kickoff(self, inputs, input_files)
+            self._log_crew_start()
 
             if self.process == Process.sequential:
                 result = self._run_sequential_process()
@@ -969,6 +970,7 @@ class Crew(FlowTrackable, BaseModel):
 
             self.usage_metrics = self.calculate_usage_metrics()
 
+            self._log_crew_finish()
             return result
         except Exception as e:
             crewai_event_bus.emit(
@@ -1464,16 +1466,25 @@ class Crew(FlowTrackable, BaseModel):
             start_index = custom_start
 
         task_outputs: list[TaskOutput] = []
-        futures: list[tuple[Task, Future[TaskOutput], int]] = []
+        futures: list[tuple[Task, Future[TaskOutput], int, str, str | None]] = []
         last_sync_output: TaskOutput | None = None
+        previous_agent_role: str | None = None
 
         for task_index, task in enumerate(tasks):
             exec_data, task_outputs, last_sync_output = prepare_task_execution(
                 self, task, task_index, start_index, task_outputs, last_sync_output
             )
             if exec_data.should_skip:
+                if exec_data.agent is None:
+                    skipped_agent = self._get_agent_to_use(task)
+                    if skipped_agent is not None:
+                        previous_agent_role = skipped_agent.role
                 continue
 
+            agent_role = exec_data.agent.role if exec_data.agent else "None"
+            if previous_agent_role and previous_agent_role != agent_role:
+                self._log_context_received(agent_role, previous_agent_role)
+
             if isinstance(task, ConditionalTask):
                 skipped_task_output = self._handle_conditional_task(
                     task, task_outputs, futures, task_index, was_replayed
@@ -1482,6 +1493,8 @@ class Crew(FlowTrackable, BaseModel):
                     task_outputs.append(skipped_task_output)
                     continue
 
+            next_agent_role = self._get_next_agent_role(tasks, task_index)
+
             if task.async_execution:
                 context = self._get_context(
                     task, [last_sync_output] if last_sync_output else []
@@ -1491,7 +1504,9 @@ class Crew(FlowTrackable, BaseModel):
                     context=context,
                     tools=exec_data.tools,
                 )
-                futures.append((task, future, task_index))
+                futures.append(
+                    (task, future, task_index, agent_role, next_agent_role)
+                )
             else:
                 if futures:
                     task_outputs.extend(
@@ -1508,6 +1523,9 @@ class Crew(FlowTrackable, BaseModel):
                 task_outputs.append(task_output)
                 self._process_task_result(task, task_output)
                 self._store_execution_log(task, task_output, task_index, was_replayed)
+                self._log_task_completion(agent_role, task, next_agent_role)
+
+            previous_agent_role = agent_role
 
         if futures:
             task_outputs.extend(self._process_async_tasks(futures, was_replayed))
@@ -1518,7 +1536,7 @@ class Crew(FlowTrackable, BaseModel):
         self,
         task: ConditionalTask,
         task_outputs: list[TaskOutput],
-        futures: list[tuple[Task, Future[TaskOutput], int]],
+        futures: list[tuple[Task, Future[TaskOutput], int, str, str | None]],
         task_index: int,
         was_replayed: bool,
     ) -> TaskOutput | None:
@@ -1738,7 +1756,29 @@ class Crew(FlowTrackable, BaseModel):
 
         return tools or []
 
+    def _log_crew_start(self) -> None:
+        crew_name = self.name or "crew"
+        self._logger.log(
+            "info",
+            f"[{crew_name}] Starting crew execution",
+            color="bold_purple",
+        )
+
+    def _log_crew_finish(self) -> None:
+        crew_name = self.name or "crew"
+        self._logger.log(
+            "info",
+            f"[{crew_name}] Crew execution completed",
+            color="bold_purple",
+        )
+
     def _log_task_start(self, task: Task, role: str = "None") -> None:
+        task_name = task.name or task.description
+        self._logger.log(
+            "info",
+            f"[Agent: {role}] Starting task: {task_name}",
+            color="bold_blue",
+        )
         if self.output_log_file:
             self._file_handler.log(
                 task_name=task.name,  # type: ignore[arg-type]
@@ -1747,6 +1787,39 @@ class Crew(FlowTrackable, BaseModel):
                 status="started",
             )
 
+    def _log_task_completion(
+        self, role: str, task: Task, next_agent_role: str | None = None
+    ) -> None:
+        task_name = task.name or task.description
+        if next_agent_role and next_agent_role != role:
+            self._logger.log(
+                "info",
+                f"[Agent: {role}] Task complete: {task_name}, passing to: {next_agent_role}",
+                color="bold_blue",
+            )
+        else:
+            self._logger.log(
+                "info",
+                f"[Agent: {role}] Task complete: {task_name}",
+                color="bold_blue",
+            )
+
+    def _log_context_received(self, role: str, from_role: str) -> None:
+        self._logger.log(
+            "info",
+            f"[Agent: {role}] Received context from {from_role}",
+            color="bold_blue",
+        )
+
+    def _get_next_agent_role(
+        self, tasks: list[Task], current_index: int
+    ) -> str | None:
+        for i in range(current_index + 1, len(tasks)):
+            next_agent = self._get_agent_to_use(tasks[i])
+            if next_agent is not None:
+                return next_agent.role
+        return None
+
     def _update_manager_tools(
         self, task: Task, tools: list[BaseTool]
     ) -> list[BaseTool]:
@@ -1818,17 +1891,18 @@ class Crew(FlowTrackable, BaseModel):
 
     def _process_async_tasks(
         self,
-        futures: list[tuple[Task, Future[TaskOutput], int]],
+        futures: list[tuple[Task, Future[TaskOutput], int, str, str | None]],
         was_replayed: bool = False,
     ) -> list[TaskOutput]:
         task_outputs: list[TaskOutput] = []
-        for future_task, future, task_index in futures:
+        for future_task, future, task_index, agent_role, next_agent_role in futures:
             task_output = future.result()
             task_outputs.append(task_output)
             self._process_task_result(future_task, task_output)
             self._store_execution_log(
                 future_task, task_output, task_index, was_replayed
             )
+            self._log_task_completion(agent_role, future_task, next_agent_role)
         return task_outputs
 
     @staticmethod

lib/crewai/tests/test_crew.py

@@ -4994,3 +4994,245 @@ def test_memory_remember_receives_task_content():
     assert "Researcher" in raw
     assert "Expected result:" in raw
     assert "Result:" in raw
+
+
+def test_verbose_crew_start_and_finish_logs(researcher, capsys):
+    """Test that crew start/finish logs are emitted when verbose=True."""
+    tasks = [
+        Task(
+            description="Research AI advancements.",
+            expected_output="A full report on AI advancements.",
+            agent=researcher,
+        ),
+    ]
+
+    crew = Crew(
+        agents=[researcher],
+        tasks=tasks,
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    mock_task_output = TaskOutput(
+        description="Research AI advancements.",
+        raw="AI advancements report",
+        agent="Researcher",
+    )
+    with patch.object(Task, "execute_sync", return_value=mock_task_output):
+        crew.kickoff()
+    captured = capsys.readouterr()
+    # Strip ANSI escape codes for assertion
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "[crew] Starting crew execution" in clean_out
+    assert "[crew] Crew execution completed" in clean_out
+
+
+def test_verbose_crew_start_and_finish_not_logged_when_not_verbose(researcher, capsys):
+    """Test that crew start/finish logs are NOT emitted when verbose=False."""
+    tasks = [
+        Task(
+            description="Research AI advancements.",
+            expected_output="A full report on AI advancements.",
+            agent=researcher,
+        ),
+    ]
+
+    crew = Crew(
+        agents=[researcher],
+        tasks=tasks,
+        process=Process.sequential,
+        verbose=False,
+    )
+
+    mock_task_output = TaskOutput(
+        description="Research AI advancements.",
+        raw="AI advancements report",
+        agent="Researcher",
+    )
+    with patch.object(Task, "execute_sync", return_value=mock_task_output):
+        crew.kickoff()
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "Starting crew execution" not in clean_out
+    assert "Crew execution completed" not in clean_out
+
+
+def test_verbose_task_start_and_completion_logs(researcher, writer, capsys):
+    """Test that per-task start/complete logs are emitted when verbose=True."""
+    tasks = [
+        Task(
+            description="Research AI advancements.",
+            expected_output="A full report on AI advancements.",
+            agent=researcher,
+        ),
+        Task(
+            description="Write about AI in healthcare.",
+            expected_output="A 4 paragraph article about AI.",
+            agent=writer,
+        ),
+    ]
+
+    crew = Crew(
+        agents=[researcher, writer],
+        tasks=tasks,
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    mock_task_output1 = TaskOutput(
+        description="Research AI advancements.",
+        raw="AI advancements report",
+        agent="Researcher",
+    )
+    mock_task_output2 = TaskOutput(
+        description="Write about AI in healthcare.",
+        raw="Article about AI in healthcare",
+        agent="Senior Writer",
+    )
+    with patch.object(
+        Task, "execute_sync", side_effect=[mock_task_output1, mock_task_output2]
+    ):
+        crew.kickoff()
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "[Agent: Researcher] Starting task:" in clean_out
+    assert "[Agent: Senior Writer] Starting task:" in clean_out
+    assert "[Agent: Researcher] Task complete:" in clean_out
+    assert "[Agent: Senior Writer] Task complete:" in clean_out
+
+
+def test_verbose_handoff_logs_between_agents(researcher, writer, capsys):
+    """Test that handoff logs appear when tasks pass between different agents."""
+    tasks = [
+        Task(
+            description="Research AI advancements.",
+            expected_output="A full report on AI advancements.",
+            agent=researcher,
+        ),
+        Task(
+            description="Write about AI in healthcare.",
+            expected_output="A 4 paragraph article about AI.",
+            agent=writer,
+        ),
+    ]
+
+    crew = Crew(
+        agents=[researcher, writer],
+        tasks=tasks,
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    mock_task_output1 = TaskOutput(
+        description="Research AI advancements.",
+        raw="AI advancements report",
+        agent="Researcher",
+    )
+    mock_task_output2 = TaskOutput(
+        description="Write about AI in healthcare.",
+        raw="Article about AI in healthcare",
+        agent="Senior Writer",
+    )
+    with patch.object(
+        Task, "execute_sync", side_effect=[mock_task_output1, mock_task_output2]
+    ):
+        crew.kickoff()
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "passing to: Senior Writer" in clean_out
+    assert "[Agent: Senior Writer] Received context from Researcher" in clean_out
+
+
+def test_verbose_no_handoff_when_same_agent(researcher, capsys):
+    """Test that no handoff log is emitted when the same agent runs consecutive tasks."""
+    tasks = [
+        Task(
+            description="Research AI advancements.",
+            expected_output="A full report on AI advancements.",
+            agent=researcher,
+        ),
+        Task(
+            description="Research more about AI.",
+            expected_output="Extended AI report.",
+            agent=researcher,
+        ),
+    ]
+
+    crew = Crew(
+        agents=[researcher],
+        tasks=tasks,
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    mock_task_output1 = TaskOutput(
+        description="Research AI advancements.",
+        raw="AI advancements report",
+        agent="Researcher",
+    )
+    mock_task_output2 = TaskOutput(
+        description="Research more about AI.",
+        raw="Extended AI report",
+        agent="Researcher",
+    )
+    with patch.object(
+        Task, "execute_sync", side_effect=[mock_task_output1, mock_task_output2]
+    ):
+        crew.kickoff()
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "Received context from" not in clean_out
+    assert "passing to:" not in clean_out
+
+
+def test_verbose_crew_name_in_logs(researcher, capsys):
+    """Test that the crew name appears in start/finish logs when set."""
+    crew = Crew(
+        name="TestCrew",
+        agents=[researcher],
+        tasks=[
+            Task(
+                description="Research AI advancements.",
+                expected_output="A full report on AI advancements.",
+                agent=researcher,
+            ),
+        ],
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    crew._log_crew_start()
+    crew._log_crew_finish()
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "[TestCrew] Starting crew execution" in clean_out
+    assert "[TestCrew] Crew execution completed" in clean_out
+
+
+def test_verbose_task_name_in_logs(researcher, writer, capsys):
+    """Test that the task name is used in logs when available, falling back to description."""
+    named_task = Task(
+        name="research_task",
+        description="Research AI advancements.",
+        expected_output="A full report on AI advancements.",
+        agent=researcher,
+    )
+    unnamed_task = Task(
+        description="Write about AI.",
+        expected_output="A short article.",
+        agent=writer,
+    )
+
+    crew = Crew(
+        agents=[researcher, writer],
+        tasks=[named_task, unnamed_task],
+        process=Process.sequential,
+        verbose=True,
+    )
+
+    crew._log_task_start(named_task, "Researcher")
+    crew._log_task_start(unnamed_task, "Senior Writer")
+    captured = capsys.readouterr()
+    clean_out = re.sub(r"\x1b\[[0-9;]*m", "", captured.out)
+    assert "[Agent: Researcher] Starting task: research_task" in clean_out
+    assert "[Agent: Senior Writer] Starting task: Write about AI." in clean_out