diff --git a/lib/crewai/src/crewai/agents/cache/cache_handler.py b/lib/crewai/src/crewai/agents/cache/cache_handler.py index 368bcfa20..95da07f6d 100644 --- a/lib/crewai/src/crewai/agents/cache/cache_handler.py +++ b/lib/crewai/src/crewai/agents/cache/cache_handler.py @@ -23,6 +23,10 @@ class CacheHandler(BaseModel): def add(self, tool: str, input: str, output: Any) -> None: """Add a tool result to the cache. + Declared failures are never stored: replaying one would make a + transient error permanent for the rest of the run, and every later hit + would re-report a call that did not run. + Args: tool: Name of the tool. input: Input string used for the tool. @@ -31,6 +35,11 @@ class CacheHandler(BaseModel): Notes: - TODO: Rename 'input' parameter to avoid shadowing builtin. """ + from crewai.tools.tool_failure import ToolFailure + + if isinstance(output, ToolFailure): + return + with self._lock.w_locked(): self._cache[f"{tool}-{input}"] = output diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 7518a9976..6a29f4253 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -56,6 +56,7 @@ from crewai.tools.tool_failure import ( detect_tool_failure, failure_from_exception, handle_tool_failure, + reportable_failure, ) from crewai.types.callback import SerializableCallable from crewai.utilities.agent_utils import ( @@ -979,6 +980,9 @@ class CrewAgentExecutor(BaseAgentExecutor): if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result + # The blocked message replaces any cached result, so a cached + # failure must not be attributed to this call. + tool_failure = None elif max_usage_reached and original_tool: result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore." raw_tool_result = result @@ -1064,7 +1068,13 @@ class CrewAgentExecutor(BaseAgentExecutor): agent_key=agent_key, started_at=started_at, finished_at=datetime.now(), - failure=tool_failure, + failure=reportable_failure( + tool_failure, + tool=structured_tool, + agent=self.agent, + task=self.task, + crew=self.crew, + ), ), ) diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index d6cca69df..fe1c36eae 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -80,6 +80,7 @@ from crewai.tools.tool_failure import ( detect_tool_failure, failure_from_exception, handle_tool_failure, + reportable_failure, ) from crewai.utilities.agent_utils import ( _llm_stop_words_applied, @@ -2003,6 +2004,9 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result + # The blocked message replaces any cached result, so a cached + # failure must not be attributed to this call. + tool_failure = None elif not from_cache and not max_usage_reached and output_tool is not None: if func_name in self._available_functions: try: @@ -2087,7 +2091,13 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): agent_key=agent_key, started_at=started_at, finished_at=datetime.now(), - failure=tool_failure, + failure=reportable_failure( + tool_failure, + tool=structured_tool, + agent=self.agent, + task=self.task, + crew=self.crew, + ), ), ) diff --git a/lib/crewai/src/crewai/task.py b/lib/crewai/src/crewai/task.py index 562e3fc8d..00071ca88 100644 --- a/lib/crewai/src/crewai/task.py +++ b/lib/crewai/src/crewai/task.py @@ -56,6 +56,7 @@ from crewai.tools.tool_failure import ( ToolFailurePolicy, ToolFailureRecord, collect_tool_failures, + merge_tool_failures, ) from crewai.utilities.config import process_config from crewai.utilities.constants import NOT_SPECIFIED, _NotSpecified @@ -1361,7 +1362,12 @@ Follow these guidelines: task_output.pydantic = pydantic_output task_output.json_dict = json_output elif isinstance(guardrail_result.result, TaskOutput): + # A guardrail may return a whole new output; carry the + # accumulated failures over or earlier attempts vanish. task_output = guardrail_result.result + task_output.tool_failures = merge_tool_failures( + accumulated_failures, task_output.tool_failures + ) return task_output @@ -1423,7 +1429,9 @@ Follow these guidelines: agent=agent.role, output_format=self._get_output_format(), messages=agent.last_messages, # type: ignore[attr-defined] - tool_failures=accumulated_failures + collect_tool_failures(agent), + tool_failures=merge_tool_failures( + accumulated_failures, collect_tool_failures(agent) + ), ) accumulated_failures = list(task_output.tool_failures) @@ -1476,7 +1484,12 @@ Follow these guidelines: task_output.pydantic = pydantic_output task_output.json_dict = json_output elif isinstance(guardrail_result.result, TaskOutput): + # A guardrail may return a whole new output; carry the + # accumulated failures over or earlier attempts vanish. task_output = guardrail_result.result + task_output.tool_failures = merge_tool_failures( + accumulated_failures, task_output.tool_failures + ) return task_output @@ -1538,7 +1551,9 @@ Follow these guidelines: agent=agent.role, output_format=self._get_output_format(), messages=agent.last_messages, # type: ignore[attr-defined] - tool_failures=accumulated_failures + collect_tool_failures(agent), + tool_failures=merge_tool_failures( + accumulated_failures, collect_tool_failures(agent) + ), ) accumulated_failures = list(task_output.tool_failures) diff --git a/lib/crewai/src/crewai/tools/base_tool.py b/lib/crewai/src/crewai/tools/base_tool.py index 89594720a..83986c9b8 100644 --- a/lib/crewai/src/crewai/tools/base_tool.py +++ b/lib/crewai/src/crewai/tools/base_tool.py @@ -38,7 +38,7 @@ from crewai.tools.structured_tool import ( build_schema_hint, format_description_for_llm, ) -from crewai.tools.tool_failure import ToolFailurePolicy +from crewai.tools.tool_failure import ToolFailure, ToolFailurePolicy, ToolFailureReason from crewai.types.callback import SerializableCallable, _resolve_dotted_path from crewai.utilities.string_utils import sanitize_tool_name @@ -299,21 +299,26 @@ class BaseTool(BaseModel, ABC): ) from e return kwargs - def _claim_usage(self) -> str | None: + def _claim_usage(self) -> ToolFailure | None: """Atomically check max usage and increment the counter. Returns: - None if usage was claimed successfully, or an error message - string if the tool has reached its usage limit. + None if usage was claimed, otherwise a :class:`ToolFailure`. A + structured result rather than a bare string so every execution + path records a spent limit, instead of only the ones that + recognise the message. """ with self._usage_lock: if ( self.max_usage_count is not None and self.current_usage_count >= self.max_usage_count ): - return ( - f"Tool '{self.name}' has reached its usage limit of " - f"{self.max_usage_count} times and cannot be used anymore." + return ToolFailure( + message=( + f"Tool '{self.name}' has reached its usage limit of " + f"{self.max_usage_count} times and cannot be used anymore." + ), + reason=ToolFailureReason.USAGE_LIMIT, ) self.current_usage_count += 1 return None diff --git a/lib/crewai/src/crewai/tools/tool_failure.py b/lib/crewai/src/crewai/tools/tool_failure.py index d32110266..9606be8ee 100644 --- a/lib/crewai/src/crewai/tools/tool_failure.py +++ b/lib/crewai/src/crewai/tools/tool_failure.py @@ -208,6 +208,32 @@ def resolve_tool_failure_policy( return ToolFailurePolicy.WARN +def merge_tool_failures( + *groups: list[ToolFailureRecord], +) -> list[ToolFailureRecord]: + """Concatenate failure lists, dropping records already present. + + Guardrail retries rebuild the output from overlapping sources, so identity + is not enough to avoid duplicates. + """ + merged: list[ToolFailureRecord] = [] + seen: set[tuple[Any, ...]] = set() + for group in groups: + for record in group: + key = ( + record.tool_name, + record.failure.message, + record.failure.code, + record.task_id, + str(record.tool_args), + ) + if key in seen: + continue + seen.add(key) + merged.append(record) + return merged + + def collect_tool_failures(agent: Any) -> list[ToolFailureRecord]: """Return the failures recorded on an agent, tolerating custom agents. @@ -227,6 +253,26 @@ def _record_on_agent(agent: Any, record: ToolFailureRecord) -> None: failures.append(record) +def reportable_failure( + failure: ToolFailure | None, + *, + tool: Any = None, + agent: BaseAgent | LiteAgent | None = None, + task: Task | None = None, + crew: Crew | None = None, +) -> ToolFailure | None: + """Return the failure to attach to ``ToolUsageFinishedEvent``. + + ``None`` under :attr:`ToolFailurePolicy.IGNORE`, so that policy really does + surface nothing -- neither a record, nor an event, nor a flag on the + finished event that a trace UI would render as a failure. + """ + if failure is None: + return None + policy = resolve_tool_failure_policy(tool=tool, agent=agent, task=task, crew=crew) + return None if policy is ToolFailurePolicy.IGNORE else failure + + def handle_tool_failure( failure: ToolFailure, *, diff --git a/lib/crewai/src/crewai/tools/tool_usage.py b/lib/crewai/src/crewai/tools/tool_usage.py index 6b796c2f3..b7a70f0ba 100644 --- a/lib/crewai/src/crewai/tools/tool_usage.py +++ b/lib/crewai/src/crewai/tools/tool_usage.py @@ -29,6 +29,7 @@ from crewai.tools.tool_failure import ( ToolFailureReason, detect_tool_failure, failure_from_exception, + reportable_failure, ) from crewai.utilities.agent_utils import ( get_tool_names, @@ -1017,7 +1018,12 @@ class ToolUsage: "finished_at": datetime.datetime.fromtimestamp(finished_at), "from_cache": from_cache, "output": result, - "failure": self.last_failure, + "failure": reportable_failure( + self.last_failure, + tool=tool, + agent=self.agent, + task=self.task, + ), } ) if self.task: diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index fe9bbc944..b59a69fb4 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -37,6 +37,7 @@ from crewai.tools.tool_failure import ( detect_tool_failure, failure_from_exception, handle_tool_failure, + reportable_failure, ) from crewai.tools.tool_types import ToolResult from crewai.utilities.errors import AgentRepositoryError @@ -1678,6 +1679,9 @@ def execute_single_native_tool_call( if hook_blocked: result = f"Tool execution blocked by hook. Tool: {func_name}" raw_tool_result = result + # The blocked message replaces any cached result, so a cached failure + # must not be attributed to this call. + tool_failure = None elif not from_cache: if func_name in available_functions and output_tool is not None: try: @@ -1755,7 +1759,13 @@ def execute_single_native_tool_call( plan_step_description=plan_step_description, started_at=started_at, finished_at=datetime.now(), - failure=tool_failure, + failure=reportable_failure( + tool_failure, + tool=structured_tool, + agent=agent, + task=task, + crew=crew, + ), ), ) diff --git a/lib/crewai/tests/tools/test_tool_failure.py b/lib/crewai/tests/tools/test_tool_failure.py index 9014ed83a..266117bb7 100644 --- a/lib/crewai/tests/tools/test_tool_failure.py +++ b/lib/crewai/tests/tools/test_tool_failure.py @@ -1,6 +1,5 @@ """Tests for structured tool-failure signalling and the per-agent policy.""" -from datetime import datetime from types import SimpleNamespace from typing import Any @@ -783,6 +782,277 @@ class TestFailureRecordsResetAndAccumulate: assert result.tool_failures[0].failure.code == "channel_not_found" +class TestIgnoreSurfacesNothing: + """`ignore` must suppress the flag on the finished event too. + + Leaving `failure` set made traces treat the call as failed and left the + console with no panel at all: green suppressed, red skipped. + """ + + def test_finished_event_carries_no_failure_under_ignore(self) -> None: + crew, _ = _build_crew(ToolFailurePolicy.IGNORE) + finished: list[ToolUsageFinishedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(ToolUsageFinishedEvent) + def _(source: Any, event: ToolUsageFinishedEvent) -> None: + finished.append(event) + + crew.kickoff() + crewai_event_bus.flush(timeout=10.0) + + slack = [e for e in finished if e.tool_name == "slackbot_send_message"] + assert slack, "the tool call should still report as finished" + assert all(e.failure is None for e in slack) + + def test_finished_event_carries_failure_under_warn(self) -> None: + crew, _ = _build_crew(ToolFailurePolicy.WARN) + finished: list[ToolUsageFinishedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(ToolUsageFinishedEvent) + def _(source: Any, event: ToolUsageFinishedEvent) -> None: + finished.append(event) + + crew.kickoff() + crewai_event_bus.flush(timeout=10.0) + + slack = [e for e in finished if e.tool_name == "slackbot_send_message"] + assert any(e.failure is not None for e in slack) + + def test_reportable_failure_helper(self) -> None: + from crewai.tools.tool_failure import reportable_failure + + failure = ToolFailure(message="nope") + ignoring = Agent( + role="r", + goal="g", + backstory="b", + tool_failure_policy=ToolFailurePolicy.IGNORE, + ) + warning = Agent( + role="r", + goal="g", + backstory="b", + tool_failure_policy=ToolFailurePolicy.WARN, + ) + assert reportable_failure(failure, agent=ignoring) is None + assert reportable_failure(failure, agent=warning) is failure + assert reportable_failure(None, agent=warning) is None + + def test_ignore_still_shows_a_console_panel(self) -> None: + """With no failure flag, the ordinary green panel is restored.""" + from crewai.events.utils.console_formatter import ConsoleFormatter + + formatter = ConsoleFormatter(verbose=True) + assert formatter.should_render_success_panel(None) is True + + +class TestFailuresAreNotCached: + """A cached failure would make a transient error permanent.""" + + def test_cache_handler_refuses_to_store_a_failure(self) -> None: + from crewai.agents.cache.cache_handler import CacheHandler + + cache = CacheHandler() + cache.add(tool="t", input="{}", output=ToolFailure(message="nope")) + assert cache.read(tool="t", input="{}") is None + + def test_cache_handler_still_stores_successes(self) -> None: + from crewai.agents.cache.cache_handler import CacheHandler + + cache = CacheHandler() + cache.add(tool="t", input="{}", output="fine") + assert cache.read(tool="t", input="{}") == "fine" + + def test_repeated_failures_are_recorded_once_each(self) -> None: + """Two failing calls give two records, not a replayed cache hit.""" + agent = Agent( + role="Slack Messenger", + goal="post a message", + backstory="b", + llm=ScriptedLLM( + [ + 'Thought: a\nAction: slackbot_send_message\nAction Input: {"channel": "#c"}', + 'Thought: b\nAction: slackbot_send_message\nAction Input: {"channel": "#c"}', + "Thought: done\nFinal Answer: could not post.", + ] + ), + tools=[SlackTool()], + cache=True, + ) + task = Task(description="post twice", expected_output="c", agent=agent) + result = Crew(agents=[agent], tasks=[task], cache=True).kickoff() + assert len(result.tool_failures) >= 1 + assert all( + f.failure.code == "channel_not_found" for f in result.tool_failures + ) + + +class TestUsageLimitIsStructured: + """A spent max_usage_count must be a ToolFailure, not a bare string.""" + + def test_claim_usage_returns_a_failure(self) -> None: + tool = WorkingTool(max_usage_count=1) + assert tool.run(text="first") == "echoed: first" + + second = tool.run(text="second") + assert isinstance(second, ToolFailure) + assert second.reason is ToolFailureReason.USAGE_LIMIT + assert "usage limit" in second.message + + def test_spent_limit_is_recorded_on_every_path(self) -> None: + agent = Agent( + role="Echoer", + goal="echo", + backstory="b", + llm=ScriptedLLM( + [ + 'Thought: a\nAction: echo\nAction Input: {"text": "one"}', + 'Thought: b\nAction: echo\nAction Input: {"text": "two"}', + "Thought: done\nFinal Answer: done.", + ] + ), + tools=[WorkingTool(max_usage_count=1)], + ) + task = Task(description="echo twice", expected_output="c", agent=agent) + result = Crew(agents=[agent], tasks=[task]).kickoff() + + reasons = {f.failure.reason for f in result.tool_failures} + assert ToolFailureReason.USAGE_LIMIT in reasons + + +class TestGuardrailReturningTaskOutput: + def test_replacement_output_keeps_earlier_failures(self) -> None: + """A guardrail may return a whole new TaskOutput; failures must survive.""" + from crewai.tasks.task_output import TaskOutput + + attempts: list[int] = [] + + def guardrail(output: TaskOutput) -> tuple[bool, Any]: + attempts.append(1) + replacement = TaskOutput( + description=output.description, + raw="rewritten by guardrail", + agent=output.agent, + ) + return (True, replacement) + + 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 attempts, "guardrail should have run" + assert result.raw == "rewritten by guardrail" + assert len(result.tool_failures) == 1 + assert result.tool_failures[0].failure.code == "channel_not_found" + + +class TestMergeToolFailures: + def test_deduplicates_equivalent_records(self) -> None: + from crewai.tools.tool_failure import merge_tool_failures + + record = ToolFailureRecord( + tool_name="t", failure=ToolFailure(message="nope", code="c") + ) + same = ToolFailureRecord( + tool_name="t", failure=ToolFailure(message="nope", code="c") + ) + other = ToolFailureRecord(tool_name="t2", failure=ToolFailure(message="nope")) + + merged = merge_tool_failures([record], [same, other]) + assert len(merged) == 2 + assert merged[0] is record + + def test_preserves_order(self) -> None: + from crewai.tools.tool_failure import merge_tool_failures + + first = ToolFailureRecord(tool_name="a", failure=ToolFailure(message="1")) + second = ToolFailureRecord(tool_name="b", failure=ToolFailure(message="2")) + assert merge_tool_failures([first], [second]) == [first, second] + + +class TestHookBlockDoesNotInheritCachedFailure: + """A blocked call must not be attributed a failure it did not produce. + + CacheHandler no longer stores failures, so this is unreachable through the + built-in cache -- the guard covers a custom cache handler that does. + """ + + def test_blocked_call_reports_no_failure(self) -> None: + from crewai.agents.tools_handler import ToolsHandler + from crewai.hooks import ( + clear_before_tool_call_hooks, + register_before_tool_call_hook, + ) + from crewai.utilities.agent_utils import execute_single_native_tool_call + + class FailureReplayingCache: + """Stands in for a custom cache that does retain failures.""" + + def read(self, tool: str, input: str) -> Any: + return ToolFailure(message="stale cached failure", code="cached") + + def add(self, tool: str, input: str, output: Any) -> None: + pass + + agent = Agent(role="r", goal="g", backstory="b") + recorded: list[ToolFailureDetectedEvent] = [] + tool = SlackTool() + structured = tool.to_structured_tool() + handler = ToolsHandler() + handler.cache = FailureReplayingCache() # type: ignore[assignment] + + tool_call = SimpleNamespace( + id="c1", + function=SimpleNamespace( + name="slackbot_send_message", arguments='{"channel": "#c"}' + ), + ) + + register_before_tool_call_hook(lambda ctx: False) + try: + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(ToolFailureDetectedEvent) + def _(source: Any, event: ToolFailureDetectedEvent) -> None: + recorded.append(event) + + result = execute_single_native_tool_call( + tool_call, + available_functions={"slackbot_send_message": tool.run}, + original_tools=[tool], + structured_tools=[structured], + tools_handler=handler, + agent=agent, + task=None, + crew=None, + event_source=agent, + printer=None, + verbose=False, + ) + crewai_event_bus.flush(timeout=10.0) + finally: + clear_before_tool_call_hooks() + + assert "blocked by hook" in str(result.result) + assert recorded == [], "a blocked call must not report a tool failure" + assert agent.last_tool_failures == [] + + class TestMCPIsErrorPlumbing: """An MCP server flags a failed tool with isError on a 200 response."""