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): scope failure accumulation per execution, drop deprecated executor

Browse Source 
Two review requests from @lorenzejay.

Accumulation no longer lives as mutable state on the shared agent. A
ContextVar collector is opened around each execution -- task, kickoff, and
each guardrail retry -- and the output reads that collector directly instead
of copying the agent's list. ContextVars are copied per asyncio task and per
thread, so concurrent executions cannot see each other's records, and
nesting is safe for retries. `last_tool_failures` prefers the active
collector and falls back to the last completed execution, so the accessor is
correct during a run too. The per-execution reset that caused the erasure is
gone.

Reproducing this took some digging and the finding is worth recording: crew
tasks *cannot* hit it, because `AgentExecutor` refuses concurrent reuse of
one instance and raises. `agent.kickoff()` has no such guard, and there the
bug reproduces exactly as reported -- two concurrent kickoffs each returned
two records. The regression test forces the overlap with a barrier so it is
deterministic rather than timing-dependent, and I verified it reports [2, 2]
against the old behavior and [1, 1] now.

Removed the tool-failure integration from `CrewAgentExecutor` entirely; that
file is back to its state on main. Note the shared ReAct helper it calls
still records failures, since that is common code rather than new behavior in
the deprecated file -- so a `raise` policy will be swallowed by that
executor's generic handler. Flagged on the PR rather than papered over.

Testing: 89 total. Two tests I wrote for this were vacuous on the first
attempt -- they passed against the simulated pre-fix code -- so each
concurrency test was checked against the old behavior before being kept.

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 10:13:33 -07:00
parent bf4ccc3f62
commit 47a37205bd
7 changed files with 317 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
lib/crewai/src/crewai/agent/core.py
View File

@@ -86,7 +86,11 @@ from crewai.skills.loader import load_skills
from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
from crewai.tools.agent_tools.agent_tools import AgentTools
from crewai.tools.tool_failure import ToolExecutionFailedError
from crewai.tools.tool_failure import (
ToolExecutionFailedError,
ToolFailureRecord,
tool_failure_collector,
)
from crewai.types.callback import SerializableCallable
from crewai.types.usage_metrics import UsageMetrics
from crewai.utilities.agent_utils import (
@@ -1464,8 +1468,6 @@ class Agent(BaseAgent):
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
@@ -1798,6 +1800,7 @@ class Agent(BaseAgent):
executor: AgentExecutor,
response_format: type[Any] | None = None,
usage_baseline: UsageMetrics | None = None,
kickoff_failures: list[ToolFailureRecord] | None = None,
) -> LiteAgentOutput:
"""Build a LiteAgentOutput from an executor result dict.
@@ -1878,7 +1881,7 @@ class Agent(BaseAgent):
todos=todo_results,
replan_count=executor.state.replan_count,
last_replan_reason=executor.state.last_replan_reason,
tool_failures=self.last_tool_failures,
tool_failures=list(kickoff_failures or []),
)
def _execute_and_build_output(
@@ -1889,9 +1892,10 @@ class Agent(BaseAgent):
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Execute the agent synchronously and build the output object."""
result = cast(dict[str, Any], executor.invoke(inputs))
with tool_failure_collector() as kickoff_failures:
result = cast(dict[str, Any], executor.invoke(inputs))
return self._build_output_from_result(
result, executor, response_format, usage_baseline
result, executor, response_format, usage_baseline, kickoff_failures
)
async def _execute_and_build_output_async(
@@ -1902,9 +1906,10 @@ class Agent(BaseAgent):
usage_baseline: UsageMetrics | None = None,
) -> LiteAgentOutput:
"""Execute the agent asynchronously and build the output object."""
result = await executor.invoke_async(inputs)
with tool_failure_collector() as kickoff_failures:
result = await executor.invoke_async(inputs)
return self._build_output_from_result(
result, executor, response_format, usage_baseline
result, executor, response_format, usage_baseline, kickoff_failures
)
def _process_kickoff_guardrail(

15
lib/crewai/src/crewai/agents/agent_builder/base_agent.py
View File

@@ -44,7 +44,11 @@ from crewai.security.security_config import SecurityConfig
from crewai.skills.models import Skill
from crewai.state.checkpoint_config import CheckpointConfig, _coerce_checkpoint
from crewai.tools.base_tool import BaseTool, Tool
from crewai.tools.tool_failure import ToolFailurePolicy, ToolFailureRecord
from crewai.tools.tool_failure import (
ToolFailurePolicy,
ToolFailureRecord,
collect_tool_failures,
)
from crewai.types.callback import SerializableCallable
from crewai.utilities.config import process_config
from crewai.utilities.i18n import I18N, get_i18n
@@ -667,10 +671,13 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
def last_tool_failures(self) -> list[ToolFailureRecord]:
"""Tool failures recorded during the most recent execution.
Empty when nothing failed or the policy is ``ignore``. Reset per
execution, like ``last_messages``. Returns a copy.
Inside an execution this reports that execution's records, so a
shared agent running concurrent tasks does not leak between them.
Outside one it reports the most recent execution, like
``last_messages``. Empty when nothing failed or the policy is
``ignore``. Returns a copy.
"""
return list(self._tool_failures)
return collect_tool_failures(self)
def reset_tool_failures(self) -> None:
"""Clear recorded tool failures before a new execution begins."""

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

@@ -49,15 +49,6 @@ from crewai.hooks.tool_hooks import (
run_after_tool_call_hooks,
run_before_tool_call_hooks,
)
from crewai.tools.tool_failure import (
ToolExecutionFailedError,
ToolFailure,
ToolFailureReason,
detect_tool_failure,
failure_from_exception,
handle_tool_failure,
reportable_failure,
)
from crewai.types.callback import SerializableCallable
from crewai.utilities.agent_utils import (
_llm_stop_words_applied,
@@ -245,9 +236,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
color="red",
)
raise
except ToolExecutionFailedError:
# A deliberate stop, not an unknown error.
raise
except Exception as e:
handle_unknown_error(PRINTER, e, verbose=self.agent.verbose)
raise
@@ -443,11 +431,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
self._invoke_step_callback(formatted_answer)
self._append_message(formatted_answer.text)
except ToolExecutionFailedError:
# A deliberate stop: the generic handler below would feed it
# back to the LLM as a recoverable observation.
raise
except OutputParserError as e:
formatted_answer = handle_output_parser_exception( # type: ignore[assignment]
e=e,
@@ -905,14 +888,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
func_args, func_name, call_id, original_tool
)
if parse_error is not None:
handle_tool_failure(
parse_error["tool_failure"],
tool_name=func_name,
tool_args=func_args,
agent=self.agent,
task=self.task,
crew=self.crew,
)
return parse_error
if original_tool is None:
@@ -950,7 +925,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
from_cache = False
result: str = "Tool not found"
raw_tool_result: Any = result
tool_failure: ToolFailure | None = None
input_str = json.dumps(args_dict) if args_dict else ""
if self.tools_handler and self.tools_handler.cache and output_tool is not None:
cached_result = self.tools_handler.cache.read(
@@ -959,7 +933,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
if cached_result is not None:
raw_tool_result = cached_result
result = format_native_tool_output_for_agent(output_tool, cached_result)
tool_failure = detect_tool_failure(cached_result)
from_cache = True
agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown"
@@ -991,15 +964,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
tool_failure = ToolFailure(
message=result, reason=ToolFailureReason.USAGE_LIMIT
)
elif (
not from_cache
and func_name in available_functions
@@ -1025,11 +992,9 @@ class CrewAgentExecutor(BaseAgentExecutor):
)
result = format_native_tool_output_for_agent(output_tool, raw_result)
tool_failure = detect_tool_failure(raw_result)
except Exception as e:
result = f"Error executing tool: {e}"
raw_tool_result = result
tool_failure = failure_from_exception(e)
if self.task:
self.task.increment_tools_errors()
crewai_event_bus.emit(
@@ -1044,14 +1009,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
),
)
error_event_emitted = True
elif not from_cache:
# Not cached and not executable: the model asked for a tool we
# do not have. The ReAct path reports this, so this one must too.
tool_failure = ToolFailure(
message=result,
reason=ToolFailureReason.UNKNOWN_TOOL,
code=func_name,
)
after_hook_context = ToolCallHookContext(
tool_name=func_name,
@@ -1079,36 +1036,15 @@ class CrewAgentExecutor(BaseAgentExecutor):
agent_key=agent_key,
started_at=started_at,
finished_at=datetime.now(),
failure=reportable_failure(
tool_failure,
tool=structured_tool,
agent=self.agent,
task=self.task,
crew=self.crew,
),
),
)
# After the finished event, so subscribers see the full lifecycle even
# when the policy aborts.
if tool_failure is not None:
handle_tool_failure(
tool_failure,
tool_name=func_name,
tool_args=args_dict,
tool=structured_tool,
agent=self.agent,
task=self.task,
crew=self.crew,
)
return {
"call_id": call_id,
"func_name": func_name,
"result": result,
"from_cache": from_cache,
"original_tool": original_tool,
"tool_failure": tool_failure,
}
def _append_tool_result_and_check_finality(
@@ -1139,8 +1075,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
original_tool
and hasattr(original_tool, "result_as_answer")
and original_tool.result_as_answer
# A failed tool must not become the final answer.
and execution_result.get("tool_failure") is None
):
return AgentFinish(
thought="Tool result is the final answer",
@@ -1180,9 +1114,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
color="red",
)
raise
except ToolExecutionFailedError:
# A deliberate stop, not an unknown error.
raise
except Exception as e:
handle_unknown_error(PRINTER, e, verbose=self.agent.verbose)
raise
@@ -1315,11 +1246,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
await self._ainvoke_step_callback(formatted_answer)
self._append_message(formatted_answer.text)
except ToolExecutionFailedError:
# A deliberate stop: the generic handler below would feed it
# back to the LLM as a recoverable observation.
raise
except OutputParserError as e:
formatted_answer = handle_output_parser_exception( # type: ignore[assignment]
e=e,

13
lib/crewai/src/crewai/lite_agent.py
View File

@@ -76,7 +76,7 @@ from crewai.tools.tool_failure import (
ToolExecutionFailedError,
ToolFailurePolicy,
ToolFailureRecord,
collect_tool_failures,
tool_failure_collector,
)
from crewai.utilities.agent_utils import (
enforce_rpm_limit,
@@ -304,6 +304,7 @@ class LiteAgent(FlowTrackable, BaseModel):
_messages: list[LLMMessage] = PrivateAttr(default_factory=list)
_iterations: int = PrivateAttr(default=0)
_tool_failures: list[ToolFailureRecord] = PrivateAttr(default_factory=list)
_kickoff_failures: list[ToolFailureRecord] = PrivateAttr(default_factory=list)
_guardrail: GuardrailCallable | None = PrivateAttr(default=None)
_guardrail_retry_count: int = PrivateAttr(default=0)
_callbacks: list[TokenCalcHandler] = PrivateAttr(default_factory=list)
@@ -549,9 +550,11 @@ class LiteAgent(FlowTrackable, BaseModel):
)
self._inject_memory_context()
return self._execute_core(
agent_info=agent_info, response_format=response_format
)
with tool_failure_collector() as kickoff_failures:
self._kickoff_failures = kickoff_failures
return self._execute_core(
agent_info=agent_info, response_format=response_format
)
except ToolExecutionFailedError as e:
# A deliberate stop, not a defect: no bug-report prompt.
@@ -733,7 +736,7 @@ class LiteAgent(FlowTrackable, BaseModel):
messages=self._messages,
# Read from whichever agent the executor was given, or the records
# go missing: original_agent under kickoff, self when standalone.
tool_failures=collect_tool_failures(self.original_agent or self),
tool_failures=list(self._kickoff_failures),
)
if self._guardrail is not None:

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

@@ -55,8 +55,8 @@ from crewai.tools.base_tool import BaseTool
from crewai.tools.tool_failure import (
ToolFailurePolicy,
ToolFailureRecord,
collect_tool_failures,
merge_tool_failures,
tool_failure_collector,
)
from crewai.utilities.config import process_config
from crewai.utilities.constants import NOT_SPECIFIED, _NotSpecified
@@ -690,11 +690,12 @@ class Task(BaseModel):
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = await agent.aexecute_task(
task=self,
context=context,
tools=tools,
)
with tool_failure_collector() as execution_failures:
result = await agent.aexecute_task(
task=self,
context=context,
tools=tools,
)
self._post_agent_execution(agent)
@@ -726,7 +727,7 @@ class Task(BaseModel):
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=list(execution_failures),
)
if self._guardrails:
@@ -845,11 +846,12 @@ class Task(BaseModel):
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = agent.execute_task(
task=self,
context=context,
tools=tools,
)
with tool_failure_collector() as execution_failures:
result = agent.execute_task(
task=self,
context=context,
tools=tools,
)
self._post_agent_execution(agent)
@@ -881,7 +883,7 @@ class Task(BaseModel):
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=list(execution_failures),
)
if self._guardrails:
@@ -1398,11 +1400,12 @@ Follow these guidelines:
content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
color="yellow",
)
result = agent.execute_task(
task=self,
context=context,
tools=tools,
)
with tool_failure_collector() as retry_failures:
result = agent.execute_task(
task=self,
context=context,
tools=tools,
)
if isinstance(result, BaseModel):
raw = result.model_dump_json()
@@ -1429,9 +1432,7 @@ Follow these guidelines:
agent=agent.role,
output_format=self._get_output_format(),
messages=agent.last_messages, # type: ignore[attr-defined]
tool_failures=merge_tool_failures(
accumulated_failures, collect_tool_failures(agent)
),
tool_failures=merge_tool_failures(accumulated_failures, retry_failures),
)
accumulated_failures = list(task_output.tool_failures)
@@ -1520,11 +1521,12 @@ Follow these guidelines:
content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
color="yellow",
)
result = await agent.aexecute_task(
task=self,
context=context,
tools=tools,
)
with tool_failure_collector() as retry_failures:
result = await agent.aexecute_task(
task=self,
context=context,
tools=tools,
)
if isinstance(result, BaseModel):
raw = result.model_dump_json()
@@ -1551,9 +1553,7 @@ Follow these guidelines:
agent=agent.role,
output_format=self._get_output_format(),
messages=agent.last_messages, # type: ignore[attr-defined]
tool_failures=merge_tool_failures(
accumulated_failures, collect_tool_failures(agent)
),
tool_failures=merge_tool_failures(accumulated_failures, retry_failures),
)
accumulated_failures = list(task_output.tool_failures)

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

@@ -12,6 +12,9 @@ declarative -- nothing here guesses whether a string "looks like" an error.
from __future__ import annotations
from collections.abc import Generator
from contextlib import contextmanager
from contextvars import ContextVar
from enum import Enum
import logging
from typing import TYPE_CHECKING, Any
@@ -235,12 +238,17 @@ def merge_tool_failures(
def collect_tool_failures(agent: Any) -> list[ToolFailureRecord]:
"""Return the failures recorded on an agent, tolerating custom agents.
"""Failures for the execution in progress, else the agent's last ones.
Third-party agents and test doubles may not expose ``last_tool_failures``
as a list, and building a task's output must never fail over telemetry.
Prefers the active collector so a shared agent running concurrent tasks
reports only the caller's own records. Tolerates agents that do not expose
the attribute at all, since reading telemetry must never raise.
"""
records = getattr(agent, "last_tool_failures", None)
active = active_tool_failures()
if active is not None:
return list(active)
records = getattr(agent, "_tool_failures", None)
if not isinstance(records, list):
return []
return [record for record in records if isinstance(record, ToolFailureRecord)]
@@ -252,8 +260,45 @@ def _agent_id(agent: Any) -> str | None:
return str(agent_id) if agent_id is not None else None
def _record_on_agent(agent: Any, record: ToolFailureRecord) -> None:
"""Append to the agent's per-execution failure list when it has one."""
_active_failures: ContextVar[list[ToolFailureRecord] | None] = ContextVar(
"crewai_tool_failures", default=None
)
@contextmanager
def tool_failure_collector() -> Generator[list[ToolFailureRecord], None, None]:
"""Collect the failures of one execution, isolated from concurrent ones.
An agent may be shared by tasks running concurrently, so accumulating on
the agent lets one execution erase or inherit another's records. The
collector is a ContextVar, which asyncio tasks and threads copy, so each
execution sees only its own. Nesting is safe: a guardrail retry can open
its own scope inside the outer one.
"""
records: list[ToolFailureRecord] = []
token = _active_failures.set(records)
try:
yield records
finally:
_active_failures.reset(token)
def active_tool_failures() -> list[ToolFailureRecord] | None:
"""Records for the execution in progress, or None outside a collector."""
return _active_failures.get()
def _record_failure(agent: Any, record: ToolFailureRecord) -> None:
"""Store a record on the active collector and on the agent.
The collector is what outputs read, so it is authoritative. The agent copy
only backs ``last_tool_failures``, which reports the most recent execution
in the same best-effort way ``last_messages`` does.
"""
records = _active_failures.get()
if records is not None:
records.append(record)
failures = getattr(agent, "_tool_failures", None)
if isinstance(failures, list):
failures.append(record)
@@ -310,7 +355,7 @@ def handle_tool_failure(
task_id=str(task.id) if task else None,
)
_record_on_agent(agent, record)
_record_failure(agent, record)
# Local import: crewai.events imports tool types back, so a module-level
# import would cycle.

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

@@ -1,5 +1,6 @@
"""Tests for structured tool-failure signalling and the per-agent policy."""
import json
from types import SimpleNamespace
from typing import Any
@@ -66,6 +67,43 @@ class ScriptedLLM(LLM):
return False
class StatelessToolLLM(LLM):
"""Calls one tool, then answers -- decided from the messages, not a counter.
Stateless so concurrent executions sharing one agent cannot interleave into
each other's script.
"""
def __new__(cls, *args: Any, **kwargs: Any) -> "StatelessToolLLM":
return object.__new__(cls)
def __init__(
self,
tool_name: str,
tool_args: dict[str, Any],
done_marker: str = "rejected the message",
) -> None:
super().__init__(model="gpt-4o")
self._tool_name = tool_name
self._tool_args = tool_args
# A sentinel from the tool's own output. Not "Observation" -- the ReAct
# prompt itself contains that word, so the stub would answer before
# ever calling the tool.
self._done_marker = done_marker
def call(self, messages, tools=None, callbacks=None, available_functions=None, **kw): # noqa: ANN001, ANN003
if self._done_marker in str(messages):
return "Thought: it failed\nFinal Answer: could not post."
return (
"Thought: posting\n"
+ f"Action: {self._tool_name}\n"
+ f"Action Input: {json.dumps(self._tool_args)}"
)
def supports_function_calling(self) -> bool:
return False
def _slack_steps() -> list[str]:
call_step = (
"Thought: posting\n"
@@ -710,17 +748,15 @@ class TestRaisePolicySurvivesEveryWrapper:
import inspect
from crewai.agent.core import Agent as AgentCls
from crewai.agents.crew_agent_executor import CrewAgentExecutor
from crewai.agents.step_executor import StepExecutor
from crewai.experimental.agent_executor import AgentExecutor
# CrewAgentExecutor is deprecated and deliberately excluded.
sites = [
(AgentCls._execute_with_timeout, "_passthrough_exceptions"),
(StepExecutor.execute, "ToolExecutionFailedError"),
(AgentExecutor.execute_tool_action, "ToolExecutionFailedError"),
(AgentExecutor.execute_native_tool, "ToolExecutionFailedError"),
(CrewAgentExecutor._invoke_loop_react, "ToolExecutionFailedError"),
(CrewAgentExecutor._ainvoke_loop_react, "ToolExecutionFailedError"),
]
for func, expected in sites:
source = inspect.getsource(func)
@@ -1176,22 +1212,6 @@ class TestFailedToolIsNotTheFinalAnswer:
assert not result.has_tool_failures
class TestDeliberateStopIsNotAnUnknownError:
def test_executor_loops_do_not_route_it_to_handle_unknown_error(self) -> None:
"""Verbose runs must not print 'An unknown error occurred' for a stop."""
import inspect
from crewai.agents.crew_agent_executor import CrewAgentExecutor
for func in (CrewAgentExecutor.invoke, CrewAgentExecutor.ainvoke):
source = inspect.getsource(func)
if "handle_unknown_error" not in source:
continue
assert "ToolExecutionFailedError" in source, (
f"{func.__qualname__} would report a deliberate stop as unknown"
)
class TestEventCarriesCorrelationIds:
"""The failure event must be correlatable with the call it describes."""
@@ -1274,6 +1294,171 @@ class TestMalformedArgumentsAreReported:
assert "INVALID_INPUT" in inspect.getsource(agent_utils.parse_tool_call_args)
class TestDeprecatedExecutorIsNotIntegrated:
"""CrewAgentExecutor is deprecated; the feature must not extend into it."""
def test_no_tool_failure_integration(self) -> None:
from pathlib import Path
import crewai
# Read the file directly: importing this module by name resolves to a
# different one in this package, so inspect would read the wrong source.
source = (
Path(crewai.__file__).parent / "agents" / "crew_agent_executor.py"
).read_text()
assert "tool_failure" not in source
assert "ToolExecutionFailedError" not in source
class TestConcurrentExecutionsAreIsolated:
"""A shared agent must not leak failures between concurrent executions.
Accumulating on the agent let one execution reset another's list and both
outputs end up with both records.
"""
@staticmethod
def _tool(channel_code: str) -> BaseTool:
class NamedSlack(BaseTool):
name: str = f"slack_{channel_code}"
description: str = "Post a message."
def _run(self, text: str) -> Any:
return ToolFailure(message=f"failed {channel_code}", code=channel_code)
return NamedSlack()
def _agent_and_task(self, code: str) -> tuple[Agent, Task]:
agent = Agent(
role=f"Poster {code}",
goal="post",
backstory="b",
llm=ScriptedLLM(
[
f'Thought: go\nAction: slack_{code}\nAction Input: {{"text": "x"}}',
"Thought: done\nFinal Answer: could not post.",
]
),
tools=[self._tool(code)],
)
task = Task(
description=f"post {code}", expected_output="c", agent=agent
)
return agent, task
def test_threads_do_not_cross_contaminate(self) -> None:
import concurrent.futures
crews = []
for code in ("aaa", "bbb", "ccc"):
agent, task = self._agent_and_task(code)
crews.append((code, Crew(agents=[agent], tasks=[task])))
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
futures = {
pool.submit(crew.kickoff): code for code, crew in crews
}
results = {
futures[f]: f.result()
for f in concurrent.futures.as_completed(futures)
}
for code, result in results.items():
codes = [f.failure.code for f in result.tool_failures]
assert codes == [code], f"{code} saw {codes}"
def test_concurrent_kickoffs_on_a_shared_agent(self) -> None:
"""The reported repro, made deterministic with a barrier.
Both kickoffs are held inside their tool call at the same time, so the
old agent-level accumulation had each reset the other's list and both
outputs came back holding two records instead of one.
Crew tasks cannot hit this -- AgentExecutor refuses concurrent reuse of
one instance -- but ``agent.kickoff()`` has no such guard.
"""
import concurrent.futures
import threading
barrier = threading.Barrier(2, timeout=30)
class BlockingFailingTool(BaseTool):
name: str = "poster"
description: str = "Post a message."
def _run(self, channel: str) -> Any:
barrier.wait()
# Phrasing the LLM stub recognises as "tool already ran".
return ToolFailure(message=f"TOOLRAN {channel}", code=channel)
agent = Agent(
role="Poster",
goal="post",
backstory="b",
llm=StatelessToolLLM("poster", {"channel": "c1"}, "TOOLRAN"),
tools=[BlockingFailingTool()],
)
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
futures = [pool.submit(agent.kickoff, f"do {n}") for n in ("A", "B")]
outputs = [f.result() for f in futures]
counts = [len(o.tool_failures) for o in outputs]
assert counts == [1, 1], f"each kickoff should hold only its own: {counts}"
def test_shared_agent_across_sequential_tasks(self) -> None:
"""One agent, two tasks: each output carries only its own record."""
agent = Agent(
role="Poster",
goal="post",
backstory="b",
llm=ScriptedLLM(_slack_steps() * 4),
tools=[SlackTool()],
)
task_a = Task(description="post A", expected_output="c", agent=agent)
task_b = Task(description="post B", expected_output="c", agent=agent)
result = Crew(agents=[agent], tasks=[task_a, task_b]).kickoff()
for task_output in result.tasks_output:
assert len(task_output.tool_failures) == 1, (
f"{task_output.name} carried {len(task_output.tool_failures)}"
)
assert len(result.tool_failures) == 2
def test_collector_is_execution_scoped(self) -> None:
from crewai.tools.tool_failure import (
active_tool_failures,
tool_failure_collector,
)
assert active_tool_failures() is None
with tool_failure_collector() as outer:
assert active_tool_failures() is outer
with tool_failure_collector() as inner:
assert active_tool_failures() is inner
assert inner is not outer
assert active_tool_failures() is outer
assert active_tool_failures() is None
@pytest.mark.asyncio
async def test_async_tasks_do_not_cross_contaminate(self) -> None:
import asyncio
crews = []
for code in ("ddd", "eee"):
agent, task = self._agent_and_task(code)
crews.append((code, Crew(agents=[agent], tasks=[task])))
outputs = await asyncio.gather(
*(crew.kickoff_async() for _, crew in crews)
)
for (code, _), result in zip(crews, outputs, strict=True):
codes = [f.failure.code for f in result.tool_failures]
assert codes == [code], f"{code} saw {codes}"
class TestMCPIsErrorPlumbing:
"""An MCP server flags a failed tool with isError on a 200 response."""