diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index fe513f08e..720077ee3 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -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( diff --git a/lib/crewai/src/crewai/agents/agent_builder/base_agent.py b/lib/crewai/src/crewai/agents/agent_builder/base_agent.py index b0bd2eb4e..49fad6aac 100644 --- a/lib/crewai/src/crewai/agents/agent_builder/base_agent.py +++ b/lib/crewai/src/crewai/agents/agent_builder/base_agent.py @@ -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.""" diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py index 1fae032d7..040520e1f 100644 --- a/lib/crewai/src/crewai/agents/crew_agent_executor.py +++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py @@ -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, diff --git a/lib/crewai/src/crewai/lite_agent.py b/lib/crewai/src/crewai/lite_agent.py index 023c74892..373b774e4 100644 --- a/lib/crewai/src/crewai/lite_agent.py +++ b/lib/crewai/src/crewai/lite_agent.py @@ -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: diff --git a/lib/crewai/src/crewai/task.py b/lib/crewai/src/crewai/task.py index 00071ca88..fb9bfb5c0 100644 --- a/lib/crewai/src/crewai/task.py +++ b/lib/crewai/src/crewai/task.py @@ -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) diff --git a/lib/crewai/src/crewai/tools/tool_failure.py b/lib/crewai/src/crewai/tools/tool_failure.py index 46c36042a..ca168ad14 100644 --- a/lib/crewai/src/crewai/tools/tool_failure.py +++ b/lib/crewai/src/crewai/tools/tool_failure.py @@ -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. diff --git a/lib/crewai/tests/tools/test_tool_failure.py b/lib/crewai/tests/tools/test_tool_failure.py index 6391a5c28..6d5518445 100644 --- a/lib/crewai/tests/tools/test_tool_failure.py +++ b/lib/crewai/tests/tools/test_tool_failure.py @@ -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."""