diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 85094d8a6..ac98c20f4 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -89,6 +89,7 @@ from crewai.tools.agent_tools.agent_tools import AgentTools from crewai.tools.tool_failure import ( ToolExecutionFailedError, ToolFailureRecord, + merge_tool_failures, tool_failure_collector, ) from crewai.types.callback import SerializableCallable @@ -1971,9 +1972,15 @@ class Agent(BaseAgent): role="user", ) - output = self._execute_and_build_output( + retried = self._execute_and_build_output( executor, inputs, response_format, usage_baseline ) + # The retry opens its own collector, so carry the blocked attempt's + # failures forward or they vanish from the final output. + retried.tool_failures = merge_tool_failures( + output.tool_failures, retried.tool_failures + ) + output = retried return self._process_kickoff_guardrail( output=output, diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 35e55bee7..54fea7ca0 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -1769,7 +1769,12 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): ordered_results[idx] = future.result() except ToolExecutionFailedError: # A deliberate stop: folding it into a tool result would - # let the remaining parallel calls carry on. + # let the remaining parallel calls carry on. Cancel the + # siblings that have not started so they never run. + # Ones already in flight cannot be interrupted -- Python + # threads are not cancellable -- so a concurrent tool may + # still complete before the abort surfaces. + pool.shutdown(wait=False, cancel_futures=True) raise except Exception as e: tool_call = runnable_tool_calls[idx] diff --git a/lib/crewai/tests/tools/test_tool_failure.py b/lib/crewai/tests/tools/test_tool_failure.py index 68917234d..a44b6916c 100644 --- a/lib/crewai/tests/tools/test_tool_failure.py +++ b/lib/crewai/tests/tools/test_tool_failure.py @@ -1588,7 +1588,7 @@ class TestBlockedCallsAreNotFailures: ) # TOOL_REPORTED is the field default, so it is produced without ever # being named; every other member has to be referenced somewhere. - for member in ToolFailureReason: + for member in list(ToolFailureReason): if member is ToolFailureReason.TOOL_REPORTED: continue assert f"ToolFailureReason.{member.name}" in sources, ( @@ -1596,6 +1596,48 @@ class TestBlockedCallsAreNotFailures: ) +class TestKickoffGuardrailRetries: + def test_blocked_attempt_failures_survive_the_retry(self) -> None: + """The retry opens its own collector, so earlier records must be merged.""" + attempts: list[int] = [] + + def guardrail(output: Any) -> tuple[bool, Any]: + attempts.append(1) + if len(attempts) == 1: + return (False, "try again") + return (True, output.raw) + + agent = Agent( + role="Slack Messenger", + goal="post", + backstory="b", + llm=StatelessToolLLM("slackbot_send_message", {"channel": "#c"}), + tools=[SlackTool()], + guardrail=guardrail, + ) + result = agent.kickoff("post it") + + assert len(attempts) == 2, "guardrail should have blocked once" + assert result.has_tool_failures + codes = [f.failure.code for f in result.tool_failures] + assert codes and all(c == "channel_not_found" for c in codes), codes + + +class TestParallelAbortCancelsPendingSiblings: + def test_pool_is_shut_down_with_cancel_futures(self) -> None: + """A pending sibling must never start once an abort is requested. + + In-flight threads cannot be interrupted in Python, so this covers the + not-yet-started ones -- the only ones that can still be prevented. + """ + import inspect + + from crewai.experimental.agent_executor import AgentExecutor + + source = inspect.getsource(AgentExecutor.execute_native_tool) + assert "cancel_futures=True" in source + + class TestKickoffResetsTheAccessor: def test_last_tool_failures_does_not_grow_across_kickoffs(self) -> None: agent = Agent(