fix: use None sentinel instead of 'unknown' in _extract_tool_name

Addresses Cursor Bugbot feedback: the string 'unknown' could collide
with a tool legitimately named 'unknown'. Now _extract_tool_name
returns None for unrecognised formats, and _should_parallelize checks
'is None' instead of '== "unknown"'.

Co-Authored-By: João <joao@crewai.com>

lib/crewai/src/crewai/cli/cli.py

@@ -27,7 +27,7 @@ from crewai.cli.tools.main import ToolCommand
 from crewai.cli.train_crew import train_crew
 from crewai.cli.triggers.main import TriggersCommand
 from crewai.cli.update_crew import update_crew
-from crewai.cli.utils import build_env_with_tool_repository_credentials, read_toml
+from crewai.cli.utils import build_env_with_all_tool_credentials, read_toml
 from crewai.memory.storage.kickoff_task_outputs_storage import (
     KickoffTaskOutputsSQLiteStorage,
 )
@@ -48,24 +48,18 @@ def crewai() -> None:
 @click.argument("uv_args", nargs=-1, type=click.UNPROCESSED)
 def uv(uv_args: tuple[str, ...]) -> None:
     """A wrapper around uv commands that adds custom tool authentication through env vars."""
-    env = os.environ.copy()
     try:
-        pyproject_data = read_toml()
-        sources = pyproject_data.get("tool", {}).get("uv", {}).get("sources", {})
-
-        for source_config in sources.values():
-            if isinstance(source_config, dict):
-                index = source_config.get("index")
-                if index:
-                    index_env = build_env_with_tool_repository_credentials(index)
-                    env.update(index_env)
-    except (FileNotFoundError, KeyError) as e:
+        # Verify pyproject.toml exists first
+        read_toml()
+    except FileNotFoundError as e:
         raise SystemExit(
             "Error. A valid pyproject.toml file is required. Check that a valid pyproject.toml file exists in the current directory."
         ) from e
     except Exception as e:
         raise SystemExit(f"Error: {e}") from e
 
+    env = build_env_with_all_tool_credentials()
+
     try:
         subprocess.run(  # noqa: S603
             ["uv", *uv_args],  # noqa: S607

lib/crewai/src/crewai/cli/install_crew.py

@@ -2,6 +2,8 @@ import subprocess
 
 import click
 
+from crewai.cli.utils import build_env_with_all_tool_credentials
+
 
 # Be mindful about changing this.
 # on some environments we don't use this command but instead uv sync directly
@@ -13,7 +15,14 @@ def install_crew(proxy_options: list[str]) -> None:
     """
     try:
         command = ["uv", "sync", *proxy_options]
-        subprocess.run(command, check=True, capture_output=False, text=True)  # noqa: S603
+
+        # Inject tool repository credentials so uv can authenticate
+        # against private package indexes (e.g. crewai tool repository).
+        # Without this, `uv sync` fails with 401 Unauthorized when the
+        # project depends on tools from a private index.
+        env = build_env_with_all_tool_credentials()
+
+        subprocess.run(command, check=True, capture_output=False, text=True, env=env)  # noqa: S603
 
     except subprocess.CalledProcessError as e:
         click.echo(f"An error occurred while running the crew: {e}", err=True)

lib/crewai/src/crewai/cli/run_crew.py

@@ -1,11 +1,10 @@
 from enum import Enum
-import os
 import subprocess
 
 import click
 from packaging import version
 
-from crewai.cli.utils import build_env_with_tool_repository_credentials, read_toml
+from crewai.cli.utils import build_env_with_all_tool_credentials, read_toml
 from crewai.cli.version import get_crewai_version
 
 
@@ -56,19 +55,7 @@ def execute_command(crew_type: CrewType) -> None:
     """
     command = ["uv", "run", "kickoff" if crew_type == CrewType.FLOW else "run_crew"]
 
-    env = os.environ.copy()
-    try:
-        pyproject_data = read_toml()
-        sources = pyproject_data.get("tool", {}).get("uv", {}).get("sources", {})
-
-        for source_config in sources.values():
-            if isinstance(source_config, dict):
-                index = source_config.get("index")
-                if index:
-                    index_env = build_env_with_tool_repository_credentials(index)
-                    env.update(index_env)
-    except Exception:  # noqa: S110
-        pass
+    env = build_env_with_all_tool_credentials()
 
     try:
         subprocess.run(command, capture_output=False, text=True, check=True, env=env)  # noqa: S603

lib/crewai/src/crewai/cli/tools/main.py

@@ -21,6 +21,7 @@ from crewai.cli.utils import (
     get_project_description,
     get_project_name,
     get_project_version,
+    read_toml,
     tree_copy,
     tree_find_and_replace,
 )
@@ -116,11 +117,26 @@ class ToolCommand(BaseCommand, PlusAPIMixin):
         self._print_tools_preview(tools_metadata)
         self._print_current_organization()
 
+        build_env = os.environ.copy()
+        try:
+            pyproject_data = read_toml()
+            sources = pyproject_data.get("tool", {}).get("uv", {}).get("sources", {})
+
+            for source_config in sources.values():
+                if isinstance(source_config, dict):
+                    index = source_config.get("index")
+                    if index:
+                        index_env = build_env_with_tool_repository_credentials(index)
+                        build_env.update(index_env)
+        except Exception:  # noqa: S110
+            pass
+
         with tempfile.TemporaryDirectory() as temp_build_dir:
             subprocess.run(  # noqa: S603
                 ["uv", "build", "--sdist", "--out-dir", temp_build_dir],  # noqa: S607
                 check=True,
                 capture_output=False,
+                env=build_env,
             )
 
             tarball_filename = next(

lib/crewai/src/crewai/cli/utils.py

@@ -484,8 +484,12 @@ def get_flows(flow_path: str = "main.py") -> list[Flow[Any]]:
             if flow_instances:
                 break
 
-    except Exception:  # noqa: S110
-        pass
+    except Exception as e:
+        import logging
+
+        logging.getLogger(__name__).debug(
+            f"Could not load tool repository credentials: {e}"
+        )
 
     return flow_instances
 
@@ -549,6 +553,31 @@ def build_env_with_tool_repository_credentials(
     return env
 
 
+def build_env_with_all_tool_credentials() -> dict[str, Any]:
+    """
+    Build environment dict with credentials for all tool repository indexes
+    found in pyproject.toml's [tool.uv.sources] section.
+
+    Returns:
+        dict: Environment variables with credentials for all private indexes.
+    """
+    env = os.environ.copy()
+    try:
+        pyproject_data = read_toml()
+        sources = pyproject_data.get("tool", {}).get("uv", {}).get("sources", {})
+
+        for source_config in sources.values():
+            if isinstance(source_config, dict):
+                index = source_config.get("index")
+                if index:
+                    index_env = build_env_with_tool_repository_credentials(index)
+                    env.update(index_env)
+    except Exception:  # noqa: S110
+        pass
+
+    return env
+
+
 @contextmanager
 def _load_module_from_file(
     init_file: Path, module_name: str | None = None

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -1661,28 +1661,39 @@ class AgentExecutor(Flow[AgentExecutorState], CrewAgentExecutorMixin):
 
         return "native_tool_completed"
 
+    def _resolve_original_tool(self, func_name: str) -> BaseTool | None:
+        """Resolve the original BaseTool from a sanitized function name.
+
+        Uses ``_tool_name_mapping`` (populated during native-tool setup) first,
+        then falls back to scanning ``original_tools`` by sanitized name.
+
+        Args:
+            func_name: The sanitized tool name to look up.
+
+        Returns:
+            The matching BaseTool, or None if not found.
+        """
+        mapping = getattr(self, "_tool_name_mapping", None)
+        if mapping and func_name in mapping:
+            mapped = mapping[func_name]
+            if isinstance(mapped, BaseTool):
+                return mapped
+        for tool in self.original_tools or []:
+            if sanitize_tool_name(tool.name) == func_name:
+                return tool
+        return None
+
     def _should_parallelize_native_tool_calls(self, tool_calls: list[Any]) -> bool:
         """Determine if native tool calls are safe to run in parallel."""
         if len(tool_calls) <= 1:
             return False
 
         for tool_call in tool_calls:
-            info = extract_tool_call_info(tool_call)
-            if not info:
+            func_name = self._extract_tool_name(tool_call)
+            if func_name is None:
                 continue
-            _, func_name, _ = info
 
-            mapping = getattr(self, "_tool_name_mapping", None)
-            original_tool: BaseTool | None = None
-            if mapping and func_name in mapping:
-                mapped = mapping[func_name]
-                if isinstance(mapped, BaseTool):
-                    original_tool = mapped
-            if original_tool is None:
-                for tool in self.original_tools or []:
-                    if sanitize_tool_name(tool.name) == func_name:
-                        original_tool = tool
-                        break
+            original_tool = self._resolve_original_tool(func_name)
 
             if not original_tool:
                 continue
@@ -1722,17 +1733,7 @@ class AgentExecutor(Flow[AgentExecutorState], CrewAgentExecutorMixin):
         # Get agent_key for event tracking
         agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown"
 
-        original_tool: BaseTool | None = None
-        mapping = getattr(self, "_tool_name_mapping", None)
-        if mapping and func_name in mapping:
-            mapped = mapping[func_name]
-            if isinstance(mapped, BaseTool):
-                original_tool = mapped
-        if original_tool is None:
-            for tool in self.original_tools or []:
-                if sanitize_tool_name(tool.name) == func_name:
-                    original_tool = tool
-                    break
+        original_tool = self._resolve_original_tool(func_name)
 
         # Check if tool has reached max usage count
         max_usage_reached = False
@@ -1907,6 +1908,42 @@ class AgentExecutor(Flow[AgentExecutorState], CrewAgentExecutorMixin):
             "original_tool": original_tool,
         }
 
+    def _extract_tool_name(self, tool_call: Any) -> str | None:
+        """Extract tool name from various tool call formats.
+
+        Returns:
+            The sanitized tool name, or ``None`` when the format is
+            unrecognised (avoids collisions with a tool legitimately
+            named "unknown").
+        """
+        if hasattr(tool_call, "function"):
+            return sanitize_tool_name(tool_call.function.name)
+        if hasattr(tool_call, "function_call") and tool_call.function_call:
+            return sanitize_tool_name(tool_call.function_call.name)
+        if hasattr(tool_call, "name"):
+            return sanitize_tool_name(tool_call.name)
+        if isinstance(tool_call, dict):
+            func_info = tool_call.get("function", {})
+            name = func_info.get("name", "") or tool_call.get("name", "")
+            return sanitize_tool_name(name) if name else None
+        return None
+
+    @router(execute_native_tool)
+    def check_native_todo_completion(
+        self,
+    ) -> Literal["todo_satisfied", "todo_not_satisfied"]:
+        """Check if the native tool execution satisfied the active todo.
+
+        Similar to check_todo_completion but for native tool execution path.
+        """
+        current_todo = self.state.todos.current_todo
+
+        if not current_todo:
+            return "todo_not_satisfied"
+
+        # For native tools, any tool execution satisfies the todo
+        return "todo_satisfied"
+
     @listen("initialized")
     def continue_iteration(self) -> Literal["check_iteration"]:
         """Bridge listener that connects iteration loop back to iteration check."""

lib/crewai/tests/agents/test_agent_executor.py

@@ -927,6 +927,30 @@ class TestNativeToolExecution:
         assert len(tool_messages) == 1
         assert tool_messages[0]["tool_call_id"] == "call_1"
 
+    def test_check_native_todo_completion_requires_current_todo(
+        self, mock_dependencies
+    ):
+        from crewai.utilities.planning_types import TodoList
+
+        executor = _build_executor(**mock_dependencies)
+
+        # No current todo → not satisfied
+        executor.state.todos = TodoList(items=[])
+        assert executor.check_native_todo_completion() == "todo_not_satisfied"
+
+        # With a current todo that has tool_to_use → satisfied
+        running = TodoItem(
+            step_number=1,
+            description="Use the expected tool",
+            tool_to_use="expected_tool",
+            status="running",
+        )
+        executor.state.todos = TodoList(items=[running])
+        assert executor.check_native_todo_completion() == "todo_satisfied"
+
+        # With a current todo without tool_to_use → still satisfied
+        running.tool_to_use = None
+        assert executor.check_native_todo_completion() == "todo_satisfied"
 
 
 class TestPlannerObserver:
@@ -2026,3 +2050,531 @@ class TestVisionImageFormatContract:
         assert hasattr(AnthropicCompletion, "_convert_image_blocks"), (
             "Anthropic provider must have _convert_image_blocks for auto-conversion"
         )
+
+
+# =========================================================================
+# Regression: PR #4656 – Tool Name Collision & Error Handling (Issue #5244)
+# =========================================================================
+
+
+class TestExtractToolName:
+    """Regression tests for _extract_tool_name across all provider formats.
+
+    Guards against the method being removed as 'unused' again (issue #5244).
+    """
+
+    @pytest.fixture
+    def mock_dependencies(self):
+        llm = Mock()
+        llm.stop = []
+        llm.supports_stop_words.return_value = True
+
+        agent = Mock()
+        agent.id = "test-agent-id"
+        agent.role = "Test Agent"
+        agent.verbose = False
+        agent.key = "test-key"
+
+        task = Mock()
+        task.description = "Test"
+        task.human_input = False
+        task.response_model = None
+
+        crew = Mock()
+        crew.verbose = False
+        crew._train = False
+
+        return {
+            "llm": llm,
+            "agent": agent,
+            "task": task,
+            "crew": crew,
+            "prompt": {"prompt": "Test"},
+            "max_iter": 10,
+            "tools": [],
+            "tools_names": "",
+            "stop_words": [],
+            "tools_description": "",
+            "tools_handler": Mock(spec=_ToolsHandler),
+        }
+
+    def test_openai_format(self, mock_dependencies):
+        """OpenAI tool call: has .function.name"""
+        executor = _build_executor(**mock_dependencies)
+        tc = Mock()
+        tc.function = Mock()
+        tc.function.name = "my_tool"
+        # Make sure hasattr checks work correctly
+        del tc.function_call
+        del tc.name
+
+        assert executor._extract_tool_name(tc) == "my_tool"
+
+    def test_gemini_format(self, mock_dependencies):
+        """Gemini tool call: has .function_call.name"""
+        executor = _build_executor(**mock_dependencies)
+        tc = Mock(spec=[])  # empty spec so hasattr returns False for unset attrs
+        tc.function_call = Mock()
+        tc.function_call.name = "gemini_tool"
+
+        assert executor._extract_tool_name(tc) == "gemini_tool"
+
+    def test_anthropic_format(self, mock_dependencies):
+        """Anthropic tool call: has .name but no .function"""
+        executor = _build_executor(**mock_dependencies)
+        tc = Mock(spec=[])
+        tc.name = "anthropic_tool"
+
+        assert executor._extract_tool_name(tc) == "anthropic_tool"
+
+    def test_dict_format_with_function(self, mock_dependencies):
+        """Dict tool call with 'function' key."""
+        executor = _build_executor(**mock_dependencies)
+        tc = {"function": {"name": "dict_tool"}, "id": "call_1"}
+
+        assert executor._extract_tool_name(tc) == "dict_tool"
+
+    def test_dict_format_with_name_only(self, mock_dependencies):
+        """Dict tool call with top-level 'name' key (Bedrock-style)."""
+        executor = _build_executor(**mock_dependencies)
+        tc = {"name": "bedrock_tool", "id": "call_1"}
+
+        assert executor._extract_tool_name(tc) == "bedrock_tool"
+
+    def test_unknown_format_returns_none(self, mock_dependencies):
+        """Completely unrecognized object returns None (not 'unknown')."""
+        executor = _build_executor(**mock_dependencies)
+
+        assert executor._extract_tool_name(42) is None
+        assert executor._extract_tool_name("not a tool call") is None
+
+    def test_sanitizes_names(self, mock_dependencies):
+        """Tool names with special characters are sanitized."""
+        executor = _build_executor(**mock_dependencies)
+        tc = Mock()
+        tc.function = Mock()
+        tc.function.name = "My Special Tool!"
+        del tc.function_call
+        del tc.name
+
+        result = executor._extract_tool_name(tc)
+        assert result == "my_special_tool"
+
+
+class TestToolNameCollisionResolution:
+    """Regression tests for tool name collision resolution via _tool_name_mapping.
+
+    Ensures that tools with duplicate sanitized names are correctly distinguished
+    and that the mapping is used throughout the execution pipeline.
+    """
+
+    def test_convert_tools_deduplicates_names(self):
+        """Two tools with the same sanitized name get unique suffixes."""
+        from crewai.utilities.agent_utils import convert_tools_to_openai_schema
+
+        tool_a = Mock()
+        tool_a.name = "search tool"
+        tool_a.description = "Tool Description: First search"
+        tool_a.args_schema = None
+        tool_a.run = Mock(return_value="a")
+
+        tool_b = Mock()
+        tool_b.name = "search tool"
+        tool_b.description = "Tool Description: Second search"
+        tool_b.args_schema = None
+        tool_b.run = Mock(return_value="b")
+
+        schemas, funcs, mapping = convert_tools_to_openai_schema([tool_a, tool_b])
+
+        assert len(schemas) == 2
+        names = [s["function"]["name"] for s in schemas]
+        # Both names must be unique
+        assert len(set(names)) == 2
+        # The second one gets a _2 suffix
+        assert names[0] == "search_tool"
+        assert names[1] == "search_tool_2"
+        # mapping correctly points to the originals
+        assert mapping["search_tool"] is tool_a
+        assert mapping["search_tool_2"] is tool_b
+        # functions are correctly mapped
+        assert funcs["search_tool"] is tool_a.run
+        assert funcs["search_tool_2"] is tool_b.run
+
+    def test_three_tools_same_name(self):
+        """Three tools with the same name get _2 and _3 suffixes."""
+        from crewai.utilities.agent_utils import convert_tools_to_openai_schema
+
+        tools = []
+        for i in range(3):
+            t = Mock()
+            t.name = "my_tool"
+            t.description = f"Tool Description: Variant {i}"
+            t.args_schema = None
+            t.run = Mock(return_value=str(i))
+            tools.append(t)
+
+        schemas, funcs, mapping = convert_tools_to_openai_schema(tools)
+
+        names = [s["function"]["name"] for s in schemas]
+        assert names == ["my_tool", "my_tool_2", "my_tool_3"]
+        assert len(set(names)) == 3
+
+
+class TestResolveOriginalTool:
+    """Tests for _resolve_original_tool which centralizes tool lookup."""
+
+    @pytest.fixture
+    def mock_dependencies(self):
+        llm = Mock()
+        llm.stop = []
+        llm.supports_stop_words.return_value = True
+
+        agent = Mock()
+        agent.id = "test-agent-id"
+        agent.role = "Test Agent"
+        agent.verbose = False
+        agent.key = "test-key"
+
+        task = Mock()
+        task.description = "Test"
+        task.human_input = False
+        task.response_model = None
+
+        crew = Mock()
+        crew.verbose = False
+        crew._train = False
+
+        return {
+            "llm": llm,
+            "agent": agent,
+            "task": task,
+            "crew": crew,
+            "prompt": {"prompt": "Test"},
+            "max_iter": 10,
+            "tools": [],
+            "tools_names": "",
+            "stop_words": [],
+            "tools_description": "",
+            "tools_handler": Mock(spec=_ToolsHandler),
+        }
+
+    def test_resolve_via_mapping(self, mock_dependencies):
+        """_resolve_original_tool finds tool via _tool_name_mapping first."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        tool = Mock(spec=BaseTool)
+        tool.name = "search_tool"
+        executor._tool_name_mapping = {"search_tool": tool}
+        executor.original_tools = []
+
+        assert executor._resolve_original_tool("search_tool") is tool
+
+    def test_resolve_via_original_tools_fallback(self, mock_dependencies):
+        """Falls back to scanning original_tools when mapping is absent."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        tool = Mock(spec=BaseTool)
+        tool.name = "search_tool"
+        executor.original_tools = [tool]
+
+        assert executor._resolve_original_tool("search_tool") is tool
+
+    def test_resolve_collision_renamed_tool(self, mock_dependencies):
+        """Collision-renamed tool (e.g. search_tool_2) found via mapping."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        tool_a = Mock(spec=BaseTool)
+        tool_a.name = "search_tool"
+        tool_b = Mock(spec=BaseTool)
+        tool_b.name = "search_tool"
+
+        executor._tool_name_mapping = {
+            "search_tool": tool_a,
+            "search_tool_2": tool_b,
+        }
+        executor.original_tools = [tool_a, tool_b]
+
+        assert executor._resolve_original_tool("search_tool") is tool_a
+        assert executor._resolve_original_tool("search_tool_2") is tool_b
+
+    def test_resolve_returns_none_for_unknown(self, mock_dependencies):
+        """Returns None when tool is not found anywhere."""
+        executor = _build_executor(**mock_dependencies)
+        executor.original_tools = []
+
+        assert executor._resolve_original_tool("nonexistent") is None
+
+
+class TestToolErrorHandlingRegression:
+    """Regression tests for tool error handling (PR #4656).
+
+    Ensures tool execution errors are injected as observations in the
+    message history and that task error counters are incremented.
+    """
+
+    @pytest.fixture
+    def mock_dependencies(self):
+        llm = Mock()
+        llm.stop = []
+        llm.supports_stop_words.return_value = True
+
+        task = Mock()
+        task.name = "Test Task"
+        task.description = "Test"
+        task.human_input = False
+        task.response_model = None
+
+        crew = Mock()
+        crew._memory = None
+        crew.verbose = False
+        crew._train = False
+
+        agent = Mock()
+        agent.id = "test-agent-id"
+        agent.role = "Test Agent"
+        agent.verbose = False
+        agent.key = "test-key"
+
+        tools_handler = Mock(spec=_ToolsHandler)
+        tools_handler.cache = None
+
+        return {
+            "llm": llm,
+            "task": task,
+            "crew": crew,
+            "agent": agent,
+            "prompt": {"prompt": "Test"},
+            "max_iter": 10,
+            "tools": [],
+            "tools_names": "",
+            "stop_words": [],
+            "tools_description": "",
+            "tools_handler": tools_handler,
+        }
+
+    def test_tool_execution_error_injected_as_observation(self, mock_dependencies):
+        """When a native tool raises an error, the error message appears in messages."""
+        executor = _build_executor(**mock_dependencies)
+
+        def failing_tool() -> str:
+            raise RuntimeError("Something went wrong")
+
+        executor._available_functions = {"failing_tool": failing_tool}
+        executor.original_tools = []
+        executor.state.pending_tool_calls = [
+            {
+                "id": "call_err",
+                "function": {"name": "failing_tool", "arguments": "{}"},
+            }
+        ]
+
+        result = executor.execute_native_tool()
+
+        assert result == "native_tool_completed"
+        tool_messages = [m for m in executor.state.messages if m.get("role") == "tool"]
+        assert len(tool_messages) == 1
+        assert "Error executing tool" in tool_messages[0]["content"]
+        assert "Something went wrong" in tool_messages[0]["content"]
+        mock_dependencies["task"].increment_tools_errors.assert_called_once()
+
+    def test_tool_execution_error_with_collision_renamed_tool(self, mock_dependencies):
+        """Error handling works correctly for collision-renamed tools."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        def failing_search() -> str:
+            raise ValueError("Search failed")
+
+        tool_obj = Mock(spec=BaseTool)
+        tool_obj.name = "search_tool"
+        tool_obj.result_as_answer = False
+        tool_obj.max_usage_count = None
+        tool_obj.current_usage_count = 0
+
+        executor._available_functions = {"search_tool_2": failing_search}
+        executor._tool_name_mapping = {"search_tool_2": tool_obj}
+        executor.original_tools = [tool_obj]
+        executor.state.pending_tool_calls = [
+            {
+                "id": "call_err2",
+                "function": {"name": "search_tool_2", "arguments": "{}"},
+            }
+        ]
+
+        result = executor.execute_native_tool()
+
+        assert result == "native_tool_completed"
+        tool_messages = [m for m in executor.state.messages if m.get("role") == "tool"]
+        assert len(tool_messages) == 1
+        assert "Search failed" in tool_messages[0]["content"]
+
+    def test_result_as_answer_with_collision_renamed_tool(self, mock_dependencies):
+        """result_as_answer works correctly for collision-renamed tools."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        tool_obj = Mock(spec=BaseTool)
+        tool_obj.name = "search_tool"
+        tool_obj.result_as_answer = True
+        tool_obj.max_usage_count = None
+        tool_obj.current_usage_count = 0
+        tool_obj.cache_function = Mock(return_value=True)
+
+        executor._available_functions = {"search_tool_2": lambda: "final answer"}
+        executor._tool_name_mapping = {"search_tool_2": tool_obj}
+        executor.original_tools = [tool_obj]
+        executor.tools = []
+        executor.state.pending_tool_calls = [
+            {
+                "id": "call_final",
+                "function": {"name": "search_tool_2", "arguments": "{}"},
+            }
+        ]
+
+        result = executor.execute_native_tool()
+
+        assert result == "tool_result_is_final"
+        assert isinstance(executor.state.current_answer, AgentFinish)
+        assert executor.state.current_answer.output == "final answer"
+
+    def test_max_usage_with_collision_renamed_tool(self, mock_dependencies):
+        """max_usage_count is respected for collision-renamed tools."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        tool_obj = Mock(spec=BaseTool)
+        tool_obj.name = "search_tool"
+        tool_obj.result_as_answer = False
+        tool_obj.max_usage_count = 2
+        tool_obj.current_usage_count = 2  # already at limit
+
+        executor._available_functions = {"search_tool_2": lambda: "should not run"}
+        executor._tool_name_mapping = {"search_tool_2": tool_obj}
+        executor.original_tools = [tool_obj]
+        executor.tools = []
+        executor.state.pending_tool_calls = [
+            {
+                "id": "call_limited",
+                "function": {"name": "search_tool_2", "arguments": "{}"},
+            }
+        ]
+
+        result = executor.execute_native_tool()
+
+        assert result == "native_tool_completed"
+        tool_messages = [m for m in executor.state.messages if m.get("role") == "tool"]
+        assert len(tool_messages) == 1
+        assert "usage limit" in tool_messages[0]["content"]
+
+
+class TestShouldParallelizeUsesExtractToolName:
+    """Regression: _should_parallelize_native_tool_calls must use _extract_tool_name.
+
+    This test class ensures the method is wired through _extract_tool_name
+    so that it cannot be removed as 'dead code' without breaking tests.
+    """
+
+    @pytest.fixture
+    def mock_dependencies(self):
+        llm = Mock()
+        llm.stop = []
+        llm.supports_stop_words.return_value = True
+
+        agent = Mock()
+        agent.id = "test-agent-id"
+        agent.role = "Test Agent"
+        agent.verbose = False
+        agent.key = "test-key"
+
+        task = Mock()
+        task.description = "Test"
+        task.human_input = False
+        task.response_model = None
+
+        crew = Mock()
+        crew.verbose = False
+        crew._train = False
+
+        return {
+            "llm": llm,
+            "agent": agent,
+            "task": task,
+            "crew": crew,
+            "prompt": {"prompt": "Test"},
+            "max_iter": 10,
+            "tools": [],
+            "tools_names": "",
+            "stop_words": [],
+            "tools_description": "",
+            "tools_handler": Mock(spec=_ToolsHandler),
+        }
+
+    def test_parallelize_calls_extract_tool_name(self, mock_dependencies):
+        """Verify _should_parallelize routes through _extract_tool_name."""
+        executor = _build_executor(**mock_dependencies)
+        executor.original_tools = []
+
+        tool_calls = [
+            {"id": "c1", "function": {"name": "tool_a", "arguments": "{}"}},
+            {"id": "c2", "function": {"name": "tool_b", "arguments": "{}"}},
+        ]
+
+        with patch.object(
+            executor, "_extract_tool_name", wraps=executor._extract_tool_name
+        ) as mock_extract:
+            executor._should_parallelize_native_tool_calls(tool_calls)
+
+        assert mock_extract.call_count == 2
+
+    def test_parallelize_false_for_result_as_answer_via_mapping(
+        self, mock_dependencies
+    ):
+        """result_as_answer tool detected via _tool_name_mapping blocks parallel."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        final_tool = Mock(spec=BaseTool)
+        final_tool.name = "search_tool"
+        final_tool.result_as_answer = True
+        final_tool.max_usage_count = None
+
+        executor._tool_name_mapping = {"search_tool_2": final_tool}
+        executor.original_tools = [final_tool]
+
+        tool_calls = [
+            {"id": "c1", "function": {"name": "search_tool_2", "arguments": "{}"}},
+            {"id": "c2", "function": {"name": "other_tool", "arguments": "{}"}},
+        ]
+
+        assert executor._should_parallelize_native_tool_calls(tool_calls) is False
+
+    def test_parallelize_true_when_all_tools_safe(self, mock_dependencies):
+        """Parallel execution allowed when no tools have special flags."""
+        from crewai.tools.base_tool import BaseTool
+
+        executor = _build_executor(**mock_dependencies)
+
+        safe_tool = Mock(spec=BaseTool)
+        safe_tool.name = "safe_tool"
+        safe_tool.result_as_answer = False
+        safe_tool.max_usage_count = None
+
+        executor._tool_name_mapping = {"safe_tool": safe_tool}
+        executor.original_tools = [safe_tool]
+
+        tool_calls = [
+            {"id": "c1", "function": {"name": "safe_tool", "arguments": "{}"}},
+            {"id": "c2", "function": {"name": "unknown_tool", "arguments": "{}"}},
+        ]
+
+        assert executor._should_parallelize_native_tool_calls(tool_calls) is True

lib/crewai/tests/cli/tools/test_main.py

@@ -218,6 +218,7 @@ def test_publish_when_not_in_sync_and_force(
         ["uv", "build", "--sdist", "--out-dir", unittest.mock.ANY],
         check=True,
         capture_output=False,
+        env=unittest.mock.ANY,
     )
     mock_open.assert_called_with(unittest.mock.ANY, "rb")
     mock_publish.assert_called_with(
@@ -279,6 +280,7 @@ def test_publish_success(
         ["uv", "build", "--sdist", "--out-dir", unittest.mock.ANY],
         check=True,
         capture_output=False,
+        env=unittest.mock.ANY,
     )
     mock_open.assert_called_with(unittest.mock.ANY, "rb")
     mock_publish.assert_called_with(