andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-08-10 08:21:54 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

fix(tools): make ignore truly silent, stop caching failures, close 4 gaps

Browse Source 
Six findings from the latest review round, all verified against the code
before touching it.

`ignore` was not silent. `ToolUsageFinishedEvent.failure` was set before
the policy ran, so traces still saw a failed call under a policy documented
as surfacing nothing. Worse, the console then showed *no* panel at all:
green was suppressed because `failure` was present, red was skipped because
`ignore` never emits `ToolFailureDetectedEvent`. New `reportable_failure()`
resolves the policy before the finished event and drops the flag under
`ignore`; wired into all four execution paths.

Failures were being cached. `CacheHandler.add` stored a `ToolFailure` like
any other result, so a transient error became permanent for the rest of the
run and every later hit re-reported a call that never re-ran. The cache now
refuses to store declared failures -- fixed at the single choke point rather
than at each of the four call sites.

A spent `max_usage_count` was invisible on the shared native path.
`BaseTool._claim_usage` returned a bare string that only the executors
recognising that exact message treated as a failure. It now returns a
`ToolFailure` with `USAGE_LIMIT`, so every path records it.

A guardrail returning a whole `TaskOutput` replaced the output without
carrying accumulated failures over, so earlier attempts vanished from
`CrewOutput.tool_failures`. New `merge_tool_failures()` combines and
deduplicates, and the retry-rebuild path uses it too.

A hook-blocked call inherited a cached failure and attributed it to a call
that never ran. Now cleared. Not reachable through the built-in cache once
failures stop being cached, so the test injects a custom cache handler that
does retain them -- verified to fail without the guard.

Also removed a `datetime` import left unused by the earlier console-test
rewrite.

Testing: 13 further tests, 73 total. Full suite matches baseline exactly at
377 pre-existing failures; the usage-limit suites that `_claim_usage`
touches pass unchanged; 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
This commit is contained in:
Joao Moura
2026-07-29 09:16:52 -07:00
parent bc99c4b98a
commit 55160c68fd
9 changed files with 395 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
lib/crewai/src/crewai/agents/cache/cache_handler.py vendored
View File

@@ -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

12
lib/crewai/src/crewai/agents/crew_agent_executor.py
View File

@@ -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,
),
),
)

12
lib/crewai/src/crewai/experimental/agent_executor.py
View File

@@ -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,
),
),
)

19
lib/crewai/src/crewai/task.py
View File

@@ -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)

19
lib/crewai/src/crewai/tools/base_tool.py
View File

@@ -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

46
lib/crewai/src/crewai/tools/tool_failure.py
View File

@@ -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,
*,

8
lib/crewai/src/crewai/tools/tool_usage.py
View File

@@ -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:

12
lib/crewai/src/crewai/utilities/agent_utils.py
View File

@@ -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,
),
),
)

272
lib/crewai/tests/tools/test_tool_failure.py
View File

@@ -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."""