fix(tools): merge failures across kickoff guardrail retries, cancel siblings

Kickoff guardrail retries discarded the blocked attempt's failures. Each
retry calls `_execute_and_build_output`, which opens a fresh collector and
builds a new output, so a run could report zero failures having demonstrably
failed one -- the same bug already fixed on the task guardrail path, which
merges. Now merged there too. Verified the test fails without the fix.

Under `raise`, one parallel native tool aborting left its siblings running:
the pool waited for them and pending ones still started. It now shuts the
pool down with `cancel_futures=True` so a not-yet-started sibling never runs.
Threads already in flight cannot be interrupted in Python, so a concurrent
tool may still complete before the abort surfaces; that is noted at the call
site rather than left implied.

Also satisfied CodeQL by materialising the enum in the guard test's loop.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

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,

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]

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(