Address review: async task logging, fix typo, use _get_agent_to_use for role

- Add completion/handoff logs for async tasks in _process_async_tasks
- Fix pre-existing filter_headeruvs typo back to filter_headers
- Use _get_agent_to_use() instead of task.agent.role in skip path
  to match the effective agent used during execution

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

src/crewai/crew.py

@@ -516,6 +516,7 @@ class Crew(BaseModel):
 
         """Starts the crew to work on its assigned tasks."""
         self._execution_span = self._telemetry.crew_execution_span(self, inputs)
+        self._log_crew_start()
         self._task_output_handler.reset()
         self._logging_color = "bold_purple"
 
@@ -562,6 +563,7 @@ class Crew(BaseModel):
         for metric in metrics:
             self.usage_metrics.add_usage_metrics(metric)
 
+        self._log_crew_finish()
         return result
 
     def kickoff_for_each(self, inputs: List[Dict[str, Any]]) -> List[CrewOutput]:
@@ -705,8 +707,9 @@ class Crew(BaseModel):
         """
 
         task_outputs: List[TaskOutput] = []
-        futures: List[Tuple[Task, Future[TaskOutput], int]] = []
+        futures: List[Tuple[Task, Future[TaskOutput], int, str, Optional[str]]] = []
         last_sync_output: Optional[TaskOutput] = None
+        previous_agent_role: Optional[str] = None
 
         for task_index, task in enumerate(tasks):
             if start_index is not None and task_index < start_index:
@@ -716,6 +719,9 @@ class Crew(BaseModel):
                     else:
                         task_outputs = [task.output]
                         last_sync_output = task.output
+                    skipped_agent = self._get_agent_to_use(task)
+                    if skipped_agent is not None:
+                        previous_agent_role = skipped_agent.role
                 continue
 
             agent_to_use = self._get_agent_to_use(task)
@@ -724,6 +730,11 @@ class Crew(BaseModel):
                     f"No agent available for task: {task.description}. Ensure that either the task has an assigned agent or a manager agent is provided."
                 )
 
+            if previous_agent_role and previous_agent_role != agent_to_use.role:
+                self._log_context_received(
+                    agent_to_use.role, previous_agent_role
+                )
+
             # Determine which tools to use - task tools take precedence over agent tools
             tools_for_task = task.tools or agent_to_use.tools or []
             tools_for_task = self._prepare_tools(
@@ -741,6 +752,8 @@ class Crew(BaseModel):
                 if 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 []
@@ -750,7 +763,9 @@ class Crew(BaseModel):
                     context=context,
                     tools=tools_for_task,
                 )
-                futures.append((task, future, task_index))
+                futures.append(
+                    (task, future, task_index, agent_to_use.role, next_agent_role)
+                )
             else:
                 if futures:
                     task_outputs = self._process_async_tasks(futures, was_replayed)
@@ -765,6 +780,11 @@ class Crew(BaseModel):
                 task_outputs = [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_to_use.role, task, next_agent_role
+                )
+
+            previous_agent_role = agent_to_use.role
 
         if futures:
             task_outputs = self._process_async_tasks(futures, was_replayed)
@@ -775,7 +795,7 @@ class Crew(BaseModel):
         self,
         task: ConditionalTask,
         task_outputs: List[TaskOutput],
-        futures: List[Tuple[Task, Future[TaskOutput], int]],
+        futures: List[Tuple[Task, Future[TaskOutput], int, str, Optional[str]]],
         task_index: int,
         was_replayed: bool,
     ) -> Optional[TaskOutput]:
@@ -859,12 +879,67 @@ class Crew(BaseModel):
             tools = self._inject_delegation_tools(tools, task.agent, agents_for_delegation)
         return tools
 
+    def _log_crew_start(self):
+        crew_name = self.name or "Crew"
+        self._logger.log(
+            "info",
+            f"[{crew_name}] Starting crew execution",
+            color="bold_purple",
+        )
+
+    def _log_crew_finish(self):
+        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"):
+        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, task=task.description, agent=role, status="started"
             )
 
+    def _log_task_completion(
+        self, role: str, task: Task, next_agent_role: Optional[str] = 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):
+        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
+    ) -> Optional[str]:
+        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[Tool]):
         if self.manager_agent:
             if task.agent:
@@ -912,17 +987,20 @@ class Crew(BaseModel):
 
     def _process_async_tasks(
         self,
-        futures: List[Tuple[Task, Future[TaskOutput], int]],
+        futures: List[Tuple[Task, Future[TaskOutput], int, str, Optional[str]]],
         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
 
     def _find_task_index(

tests/crew_test.py

@@ -3125,4 +3125,238 @@ def test_multimodal_agent_live_image_analysis():
     # Verify we got a meaningful response
     assert isinstance(result.raw, str)
     assert len(result.raw) > 100  # Expecting a detailed analysis
-    assert "error" not in result.raw.lower()  # No error messages in response
+    assert "error" not in result.raw.lower()  # No error messages in response
+
+
+def test_verbose_crew_start_and_finish_logs(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()
+    assert "[Crew] Starting crew execution" in captured.out
+    assert "[Crew] Crew execution completed" in captured.out
+
+
+def test_verbose_crew_start_and_finish_not_logged_when_not_verbose(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()
+    assert "[Crew] Starting crew execution" not in captured.out
+    assert "[Crew] Crew execution completed" not in captured.out
+
+
+def test_verbose_task_start_and_completion_logs(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()
+    assert "[Agent: Researcher] Starting task:" in captured.out
+    assert "[Agent: Senior Writer] Starting task:" in captured.out
+    assert "[Agent: Researcher] Task complete:" in captured.out
+    assert "[Agent: Senior Writer] Task complete:" in captured.out
+
+
+def test_verbose_handoff_logs_between_agents(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()
+    assert "passing to: Senior Writer" in captured.out
+    assert "[Agent: Senior Writer] Received context from Researcher" in captured.out
+
+
+def test_verbose_no_handoff_when_same_agent(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()
+    assert "Received context from" not in captured.out
+    assert "passing to:" not in captured.out
+
+
+def test_verbose_crew_name_in_logs(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()
+    assert "[TestCrew] Starting crew execution" in captured.out
+    assert "[TestCrew] Crew execution completed" in captured.out
+
+
+def test_verbose_task_name_in_logs(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()
+    assert "[Agent: Researcher] Starting task: research_task" in captured.out
+    assert "[Agent: Senior Writer] Starting task: Write about AI." in captured.out