fix(tools): address review round 2 and fix CI type failure

CI caught a type error I should have: widening `agent` to accept a
`LiteAgent` (so a standalone LiteAgent resolves its own policy) left the
declared signatures behind. Widened `execute_tool_and_check_finality`,
its async twin, and `ToolCallHookContext` to `Agent | BaseAgent |
LiteAgent | None`, which is what those actually receive now.

Seven CodeRabbit findings, all verified against the code first:

`raise` was being downgraded by three enclosing handlers. With
`max_execution_time` set, `_execute_with_timeout` wrapped every exception
in `RuntimeError`, so `_check_execution_error` no longer recognized the
passthrough and sent the task through the retry loop instead of aborting.
`StepExecutor.execute` turned it into `StepResult(success=False)` and let
the plan continue. `LiteAgent.kickoff` ran it through
`handle_unknown_error` and printed "This is likely a bug - please report
it" for what is a deliberate, configured stop.

Failure records were dropped on two paths. `reset_tool_failures()` only
ran in `_prepare_task_execution`, so `Agent.kickoff()` / `kickoff_async()`
— which enter through `_prepare_kickoff` — accumulated records across
runs. And a guardrail retry calls `execute_task` again, which resets the
agent, so a tool that failed on a blocked attempt vanished from the final
output entirely: a run could report zero failures having demonstrably
failed one. Failures now accumulate across guardrail attempts.

Writing the tests for that surfaced a further miss of my own:
`Agent.kickoff()` builds its `LiteAgentOutput` in `agent/core.py` via
`AgentExecutor`, not through `LiteAgent`, so `tool_failures` was always
empty there regardless of the recording fix. Wired up, and the LiteAgent
path now reads from whichever agent the executor was handed
(`original_agent` under kickoff, `self` standalone) rather than assuming.

`last_tool_failures` returns a copy, so a caller cannot mutate the
agent's record or watch it shift mid-run.

Testing: 7 further tests, 52 total, covering the timeout wrapper, the
retry limit, kickoff reset, the kickoff output path, copy semantics and
guardrail accumulation. Full suite matches baseline exactly at 377
pre-existing failures; mypy clean on every changed file.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

Author: joaomdmoura

lib/crewai/src/crewai/agent/core.py

@@ -920,6 +920,13 @@ def _execute_with_timeout(self, task_prompt: str, task: Task, timeout: int) -> A
                 raise TimeoutError(
                     f"Task '{task.description}' execution timed out after {timeout} seconds. Consider increasing max_execution_time or optimizing the task."
                 ) from e
+            except _passthrough_exceptions:
+                # A deliberate stop (e.g. tool_failure_policy="raise") must
+                # keep its type: wrapping it in RuntimeError would hide it from
+                # _check_execution_error and send the task through the retry
+                # loop instead of aborting.
+                future.cancel()
+                raise
             except Exception as e:
                 future.cancel()
                 raise RuntimeError(f"Task execution failed: {e!s}") from e
@@ -1460,6 +1467,8 @@ def _prepare_kickoff(
         Returns:
             Tuple of (executor, inputs, agent_info, parsed_tools) ready for execution.
         """
+        self.reset_tool_failures()
+
         if self.tools_handler:
             self.tools_handler.last_used_tool = None
 
@@ -1872,6 +1881,7 @@ def _build_output_from_result(
             todos=todo_results,
             replan_count=executor.state.replan_count,
             last_replan_reason=executor.state.last_replan_reason,
+            tool_failures=self.last_tool_failures,
         )
 
     def _execute_and_build_output(

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -671,10 +671,13 @@ def last_tool_failures(self) -> list[ToolFailureRecord]:
         """Tool failures recorded during the most recent execution.
 
         Empty when nothing failed, or when ``tool_failure_policy`` is
-        ``ignore``. Reset at the start of each task execution, mirroring
-        ``last_messages``.
+        ``ignore``. Reset at the start of each task execution or kickoff,
+        mirroring ``last_messages``.
+
+        Returns a copy, so a caller holding the list cannot mutate the
+        agent's record or watch it change under them mid-run.
         """
-        return self._tool_failures
+        return list(self._tool_failures)
 
     def reset_tool_failures(self) -> None:
         """Clear recorded tool failures before a new execution begins."""

lib/crewai/src/crewai/agents/step_executor.py

@@ -28,6 +28,7 @@
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
 )
+from crewai.tools.tool_failure import ToolExecutionFailedError
 from crewai.utilities.agent_utils import (
     build_text_tool_calling_fallback_message,
     build_tool_calls_assistant_message,
@@ -180,6 +181,11 @@ def execute(
                 tool_calls_made=tool_calls_made,
                 execution_time=elapsed,
             )
+        except ToolExecutionFailedError:
+            # tool_failure_policy="raise" asked for the run to stop; turning it
+            # into StepResult(success=False) would let the plan carry on.
+            raise
+
         except Exception as e:
             if self._use_native_tools and is_native_tool_calling_unsupported_error(e):
                 try:

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -23,6 +23,7 @@
     from crewai.agent import Agent
     from crewai.agents.agent_builder.base_agent import BaseAgent
     from crewai.crew import Crew
+    from crewai.lite_agent import LiteAgent
     from crewai.task import Task
     from crewai.tools.structured_tool import CrewStructuredTool
 
@@ -55,7 +56,7 @@ def __init__(
         tool_name: str,
         tool_input: dict[str, Any],
         tool: CrewStructuredTool,
-        agent: Agent | BaseAgent | None = None,
+        agent: Agent | BaseAgent | LiteAgent | None = None,
         task: Task | None = None,
         crew: Crew | None = None,
         tool_result: str | None = None,

lib/crewai/src/crewai/lite_agent.py

@@ -76,6 +76,7 @@
     ToolExecutionFailedError,
     ToolFailurePolicy,
     ToolFailureRecord,
+    collect_tool_failures,
 )
 from crewai.utilities.agent_utils import (
     enforce_rpm_limit,
@@ -463,6 +464,15 @@ def _original_role(self) -> str:
         """Return the original role for compatibility with tool interfaces."""
         return self.role
 
+    @property
+    def last_tool_failures(self) -> list[ToolFailureRecord]:
+        """Tool failures recorded during the most recent kickoff.
+
+        Same name and meaning as ``BaseAgent.last_tool_failures``, so the
+        shared collection helper works for a standalone LiteAgent too.
+        """
+        return list(self._tool_failures)
+
     @property
     def before_llm_call_hooks(
         self,
@@ -543,6 +553,23 @@ def kickoff(
                 agent_info=agent_info, response_format=response_format
             )
 
+        except ToolExecutionFailedError as e:
+            # A deliberate stop, not a defect: do not tell the user to file a
+            # bug, and do not run it through handle_unknown_error.
+            if self.verbose:
+                PRINTER.print(
+                    content=f"Agent stopped: {e}",
+                    color="red",
+                )
+            crewai_event_bus.emit(
+                self,
+                event=LiteAgentExecutionErrorEvent(
+                    agent_info=agent_info,
+                    error=str(e),
+                ),
+            )
+            raise
+
         except Exception as e:
             if self.verbose:
                 PRINTER.print(
@@ -705,7 +732,11 @@ def _execute_core(
             agent_role=self.role,
             usage_metrics=usage_metrics.model_dump() if usage_metrics else None,
             messages=self._messages,
-            tool_failures=list(self._tool_failures),
+            # Failures are recorded against whichever agent the executor was
+            # given, which is ``original_agent`` on the Agent.kickoff() path
+            # and ``self`` for a standalone LiteAgent. Read from the same one
+            # or the records go missing from the output.
+            tool_failures=collect_tool_failures(self.original_agent or self),
         )
 
         if self._guardrail is not None:

lib/crewai/src/crewai/task.py

@@ -52,7 +52,11 @@
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
 from crewai.tools.base_tool import BaseTool
-from crewai.tools.tool_failure import ToolFailurePolicy, collect_tool_failures
+from crewai.tools.tool_failure import (
+    ToolFailurePolicy,
+    ToolFailureRecord,
+    collect_tool_failures,
+)
 from crewai.utilities.config import process_config
 from crewai.utilities.constants import NOT_SPECIFIED, _NotSpecified
 from crewai.utilities.converter import (
@@ -1330,6 +1334,11 @@ def _invoke_guardrail_function(
 
         max_attempts = self.guardrail_max_retries + 1
 
+        # Each retry calls agent.execute_task again, which resets the agent's
+        # per-execution failure list. Accumulate across attempts so a tool that
+        # failed on a blocked attempt is still reported on the final output.
+        accumulated_failures: list[ToolFailureRecord] = list(task_output.tool_failures)
+
         for attempt in range(max_attempts):
             guardrail_result = process_guardrail(
                 output=task_output,
@@ -1416,8 +1425,9 @@ def _invoke_guardrail_function(
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
-                tool_failures=collect_tool_failures(agent),
+                tool_failures=accumulated_failures + collect_tool_failures(agent),
             )
+            accumulated_failures = list(task_output.tool_failures)
 
         return task_output
 
@@ -1440,6 +1450,11 @@ async def _ainvoke_guardrail_function(
 
         max_attempts = self.guardrail_max_retries + 1
 
+        # Each retry calls agent.execute_task again, which resets the agent's
+        # per-execution failure list. Accumulate across attempts so a tool that
+        # failed on a blocked attempt is still reported on the final output.
+        accumulated_failures: list[ToolFailureRecord] = list(task_output.tool_failures)
+
         for attempt in range(max_attempts):
             guardrail_result = process_guardrail(
                 output=task_output,
@@ -1526,7 +1541,8 @@ async def _ainvoke_guardrail_function(
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
-                tool_failures=collect_tool_failures(agent),
+                tool_failures=accumulated_failures + collect_tool_failures(agent),
             )
+            accumulated_failures = list(task_output.tool_failures)
 
         return task_output

lib/crewai/src/crewai/utilities/tool_utils.py

@@ -26,6 +26,7 @@
     from crewai.agent import Agent
     from crewai.agents.agent_builder.base_agent import BaseAgent
     from crewai.crew import Crew
+    from crewai.lite_agent import LiteAgent
     from crewai.llm import LLM
     from crewai.llms.base_llm import BaseLLM
     from crewai.task import Task
@@ -38,7 +39,7 @@ async def aexecute_tool_and_check_finality(
     agent_role: str | None = None,
     tools_handler: ToolsHandler | None = None,
     task: Task | None = None,
-    agent: Agent | BaseAgent | None = None,
+    agent: Agent | BaseAgent | LiteAgent | None = None,
     function_calling_llm: BaseLLM | LLM | None = None,
     fingerprint_context: dict[str, str] | None = None,
     crew: Crew | None = None,
@@ -187,7 +188,7 @@ def execute_tool_and_check_finality(
     agent_role: str | None = None,
     tools_handler: ToolsHandler | None = None,
     task: Task | None = None,
-    agent: Agent | BaseAgent | None = None,
+    agent: Agent | BaseAgent | LiteAgent | None = None,
     function_calling_llm: BaseLLM | LLM | None = None,
     fingerprint_context: dict[str, str] | None = None,
     crew: Crew | None = None,

lib/crewai/tests/tools/test_tool_failure.py

@@ -562,6 +562,125 @@ def test_has_tool_failures_exists_on_all_output_types(self) -> None:
         assert CrewOutput().has_tool_failures is False
 
 
+class TestRaisePolicySurvivesEveryWrapper:
+    """`raise` must abort, not get downgraded by an enclosing handler."""
+
+    def test_timeout_wrapper_preserves_the_error_type(self) -> None:
+        """max_execution_time wraps failures in RuntimeError; not this one."""
+        agent = Agent(
+            role="Slack Messenger",
+            goal="post a message",
+            backstory="b",
+            llm=ScriptedLLM(_slack_steps()),
+            tools=[SlackTool()],
+            tool_failure_policy=ToolFailurePolicy.RAISE,
+            max_execution_time=30,
+        )
+        task = Task(description="post to slack", expected_output="c", agent=agent)
+
+        with pytest.raises(ToolExecutionFailedError):
+            Crew(agents=[agent], tasks=[task]).kickoff()
+
+    def test_retry_limit_does_not_swallow_the_abort(self) -> None:
+        """A deliberate stop must not be retried as a transient error."""
+        agent = Agent(
+            role="Slack Messenger",
+            goal="post a message",
+            backstory="b",
+            llm=ScriptedLLM(_slack_steps()),
+            tools=[SlackTool()],
+            tool_failure_policy=ToolFailurePolicy.RAISE,
+            max_retry_limit=3,
+        )
+        task = Task(description="post to slack", expected_output="c", agent=agent)
+
+        with pytest.raises(ToolExecutionFailedError):
+            Crew(agents=[agent], tasks=[task]).kickoff()
+        assert agent._times_executed == 0, "the abort must not trigger retries"
+
+    def test_passthrough_tuple_includes_the_error(self) -> None:
+        from crewai.agent.core import _passthrough_exceptions
+
+        assert ToolExecutionFailedError in _passthrough_exceptions
+
+
+class TestFailureRecordsResetAndAccumulate:
+    def test_kickoff_resets_between_runs(self) -> None:
+        """Agent.kickoff() goes through _prepare_kickoff, not task execution."""
+        agent = Agent(
+            role="Slack Messenger",
+            goal="post a message",
+            backstory="b",
+            llm=ScriptedLLM(_slack_steps()),
+            tools=[SlackTool()],
+        )
+
+        first = agent.kickoff("post it")
+        assert len(first.tool_failures) == 1
+        assert first.has_tool_failures
+
+        agent.llm = ScriptedLLM(_slack_steps())
+        second = agent.kickoff("post it again")
+        assert len(second.tool_failures) == 1, "records must not accumulate"
+
+    def test_kickoff_output_sees_failures_recorded_on_the_agent(self) -> None:
+        """The LiteAgent under kickoff records against the owning Agent."""
+        agent = Agent(
+            role="Slack Messenger",
+            goal="post a message",
+            backstory="b",
+            llm=ScriptedLLM(_slack_steps()),
+            tools=[SlackTool()],
+        )
+        result = agent.kickoff("post it")
+        assert [f.failure.code for f in result.tool_failures] == ["channel_not_found"]
+
+    def test_last_tool_failures_returns_a_copy(self) -> None:
+        agent = Agent(role="r", goal="g", backstory="b")
+        agent._tool_failures.append(
+            ToolFailureRecord(tool_name="t", failure=ToolFailure(message="nope"))
+        )
+        snapshot = agent.last_tool_failures
+        snapshot.clear()
+        assert len(agent.last_tool_failures) == 1
+
+    def test_guardrail_retry_preserves_earlier_failures(self) -> None:
+        """A blocked attempt's failures must survive into the final output.
+
+        The retry calls ``agent.execute_task`` again, which resets the agent's
+        record. Without accumulation this output would report zero failures
+        even though a tool demonstrably failed on the first attempt.
+        """
+        attempts: list[int] = []
+
+        def guardrail(output: Any) -> tuple[bool, Any]:
+            attempts.append(1)
+            if len(attempts) == 1:
+                return (False, "needs another pass")
+            return (True, output.raw)
+
+        agent = Agent(
+            role="Slack Messenger",
+            goal="post a message",
+            backstory="b",
+            llm=ScriptedLLM(_slack_steps()),
+            tools=[SlackTool()],
+        )
+        task = Task(
+            description="post to slack",
+            expected_output="c",
+            agent=agent,
+            guardrail=guardrail,
+        )
+        result = Crew(agents=[agent], tasks=[task]).kickoff()
+
+        assert len(attempts) == 2, "guardrail should have blocked once"
+        # The scripted LLM answers directly on the retry, so the single
+        # surviving record is the one from the blocked first attempt.
+        assert len(result.tool_failures) == 1
+        assert result.tool_failures[0].failure.code == "channel_not_found"
+
+
 class TestMCPIsErrorPlumbing:
     """An MCP server flags a failed tool with isError on a 200 response."""