crewAI/lib/crewai/tests/tools/test_tool_usage.py

import datetime
from collections.abc import Callable
import json
import random
import threading
import time
from unittest.mock import MagicMock, patch

from crewai import Agent, Task
from crewai.agents.cache.cache_handler import CacheHandler
from crewai.agents.parser import AgentAction
from crewai.agents.tools_handler import ToolsHandler
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.tool_usage_events import (
    ToolSelectionErrorEvent,
    ToolUsageErrorEvent,
    ToolUsageFinishedEvent,
    ToolUsageStartedEvent,
    ToolValidateInputErrorEvent,
)
from crewai.hooks.tool_hooks import (
    ToolCallHookContext,
    clear_after_tool_call_hooks,
    register_after_tool_call_hook,
)
from crewai.tools import BaseTool
from crewai.tools.tool_calling import ToolCalling
from crewai.tools.tool_usage import ToolUsage
from crewai.utilities.tool_utils import execute_tool_and_check_finality
from pydantic import BaseModel, Field
import pytest


class RandomNumberToolInput(BaseModel):
    min_value: int = Field(
        ..., description="The minimum value of the range (inclusive)"
    )
    max_value: int = Field(
        ..., description="The maximum value of the range (inclusive)"
    )


class RandomNumberTool(BaseTool):
    name: str = "Random Number Generator"
    description: str = "Generates a random number within a specified range"
    args_schema: type[BaseModel] = RandomNumberToolInput

    def _run(self, min_value: int, max_value: int) -> int:
        return random.randint(min_value, max_value)  # noqa: S311


class SearchOutput(BaseModel):
    query: str
    score: float


class TypedSearchTool(BaseTool):
    name: str = "typed_search"
    description: str = "Search for a query"

    def _run(self, query: str) -> SearchOutput:
        return SearchOutput(query=query, score=0.7)


# Example agent and task
example_agent = Agent(
    role="Number Generator",
    goal="Generate random numbers for various purposes",
    backstory="You are an AI agent specialized in generating random numbers within specified ranges.",
    tools=[RandomNumberTool()],
    verbose=True,
)

example_task = Task(
    description="Generate a random number between 1 and 100",
    expected_output="A random number between 1 and 100",
    agent=example_agent,
)


def test_random_number_tool_range():
    tool = RandomNumberTool()
    result = tool._run(1, 10)
    assert 1 <= result <= 10


def test_random_number_tool_invalid_range():
    tool = RandomNumberTool()
    with pytest.raises(ValueError):
        tool._run(10, 1)


def test_random_number_tool_schema():
    tool = RandomNumberTool()

    schema = tool.args_schema.model_json_schema()

    schema_str = json.dumps(schema)

    assert "min_value" in schema_str
    assert "max_value" in schema_str

    schema_dict = json.loads(schema_str)

    assert schema_dict["properties"]["min_value"]["type"] == "integer"
    assert schema_dict["properties"]["max_value"]["type"] == "integer"

    assert (
        "minimum value" in schema_dict["properties"]["min_value"]["description"].lower()
    )
    assert (
        "maximum value" in schema_dict["properties"]["max_value"]["description"].lower()
    )


def test_tool_usage_render():
    tool = RandomNumberTool()

    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[tool],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    rendered = tool_usage._render()

    assert "Tool Name: random_number_generator" in rendered
    assert "Tool Arguments:" in rendered
    assert (
        "Tool Description: Generates a random number within a specified range"
        in rendered
    )

    assert '"min_value"' in rendered
    assert '"max_value"' in rendered
    assert '"type": "integer"' in rendered
    assert '"description": "The minimum value of the range (inclusive)"' in rendered
    assert '"description": "The maximum value of the range (inclusive)"' in rendered


def test_tool_usage_returns_json_agent_text_for_typed_output():
    tool = TypedSearchTool().to_structured_tool()
    tool_usage = ToolUsage(
        tools_handler=None,
        tools=[tool],
        task=None,
        function_calling_llm=MagicMock(),
        agent=None,
        action=MagicMock(),
    )

    result = tool_usage.use(
        calling=ToolCalling(
            tool_name="typed_search",
            arguments={"query": "crew"},
        ),
        tool_string='Action: typed_search\nAction Input: {"query": "crew"}',
    )

    assert json.loads(result) == {"query": "crew", "score": 0.7}


def test_tool_usage_cache_callback_receives_raw_typed_output():
    raw_results: list[object] = []

    def cache_result(_args: object, result: object) -> bool:
        raw_results.append(result)
        return True

    class CacheAwareTypedSearchTool(TypedSearchTool):
        cache_function: Callable = cache_result

    tools_handler = MagicMock()
    tools_handler.cache = None
    tools_handler.last_used_tool = None
    tool = CacheAwareTypedSearchTool().to_structured_tool()
    tool_usage = ToolUsage(
        tools_handler=tools_handler,
        tools=[tool],
        task=None,
        function_calling_llm=MagicMock(),
        agent=None,
        action=MagicMock(),
    )

    result = tool_usage.use(
        calling=ToolCalling(
            tool_name="typed_search",
            arguments={"query": "crew"},
        ),
        tool_string='Action: typed_search\nAction Input: {"query": "crew"}',
    )

    assert json.loads(result) == {"query": "crew", "score": 0.7}
    assert raw_results == [SearchOutput(query="crew", score=0.7)]
    tools_handler.on_tool_use.assert_called_once()
    assert tools_handler.on_tool_use.call_args.kwargs["output"] == SearchOutput(
        query="crew",
        score=0.7,
    )


def test_react_tool_hooks_receive_agent_text_and_raw_cached_typed_output():
    structured_tool = TypedSearchTool().to_structured_tool()
    tools_handler = ToolsHandler(cache=CacheHandler())
    seen_results: list[tuple[str | None, object]] = []

    def after_hook(context: ToolCallHookContext) -> None:
        seen_results.append((context.tool_result, context.raw_tool_result))

    clear_after_tool_call_hooks()
    register_after_tool_call_hook(after_hook)

    action = AgentAction(
        thought="",
        tool="typed_search",
        tool_input='{"query": "crew"}',
        text='Action: typed_search\nAction Input: {"query": "crew"}',
    )

    try:
        first = execute_tool_and_check_finality(
            agent_action=action,
            tools=[structured_tool],
            tools_handler=tools_handler,
        )
        tools_handler.last_used_tool = None
        second = execute_tool_and_check_finality(
            agent_action=action,
            tools=[structured_tool],
            tools_handler=tools_handler,
        )
    finally:
        clear_after_tool_call_hooks()

    assert json.loads(first.result) == {"query": "crew", "score": 0.7}
    assert json.loads(second.result) == {"query": "crew", "score": 0.7}
    assert seen_results == [
        ('{"query":"crew","score":0.7}', SearchOutput(query="crew", score=0.7)),
        ('{"query":"crew","score":0.7}', SearchOutput(query="crew", score=0.7)),
    ]


def test_last_raw_result_falls_back_only_until_recorded():
    tool_usage = ToolUsage(
        tools_handler=None,
        tools=[],
        task=None,
        function_calling_llm=MagicMock(),
        agent=None,
        action=MagicMock(),
    )

    assert tool_usage.get_last_raw_result("formatted result") == "formatted result"

    tool_usage.last_raw_result = None

    assert tool_usage.get_last_raw_result("formatted result") is None


def test_validate_tool_input_booleans_and_none():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"key1": True, "key2": False, "key3": None}'
    expected_arguments = {"key1": True, "key2": False, "key3": None}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_mixed_types():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"number": 123, "text": "Some text", "flag": True}'
    expected_arguments = {"number": 123, "text": "Some text", "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_single_quotes():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = "{'key': 'value', 'flag': True}"
    expected_arguments = {"key": "value", "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_invalid_json_repairable():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"key": "value", "list": [1, 2, 3,]}'
    expected_arguments = {"key": "value", "list": [1, 2, 3]}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_with_special_characters():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=MagicMock(),
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"message": "Hello, world! \u263a", "valid": True}'
    expected_arguments = {"message": "Hello, world! ☺", "valid": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_none_input():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    arguments = tool_usage._validate_tool_input(None)
    assert arguments == {}


def test_validate_tool_input_valid_json():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"key": "value", "number": 42, "flag": true}'
    expected_arguments = {"key": "value", "number": 42, "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_python_dict():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = "{'key': 'value', 'number': 42, 'flag': True}"
    expected_arguments = {"key": "value", "number": 42, "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_json5_unquoted_keys():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = "{key: 'value', number: 42, flag: true}"
    expected_arguments = {"key": "value", "number": 42, "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_with_trailing_commas():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"key": "value", "number": 42, "flag": true,}'
    expected_arguments = {"key": "value", "number": 42, "flag": True}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_invalid_input():
    mock_agent = MagicMock()
    mock_agent.key = "test_agent_key"
    mock_agent.role = "test_agent_role"
    mock_agent._original_role = "test_agent_role"
    mock_agent.verbose = False

    mock_action = MagicMock()
    mock_action.tool = "test_tool"
    mock_action.tool_input = "test_input"

    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=mock_agent,
        action=mock_action,
    )

    invalid_inputs = [
        "Just a string",
        "['list', 'of', 'values']",
        "12345",
        "",
    ]

    for invalid_input in invalid_inputs:
        with pytest.raises(Exception) as e_info:
            tool_usage._validate_tool_input(invalid_input)
        assert (
            "Tool input must be a valid dictionary in JSON or Python literal format"
            in str(e_info.value)
        )

    arguments = tool_usage._validate_tool_input(None)
    assert arguments == {}


def test_validate_tool_input_complex_structure():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = """
    {
        "user": {
            "name": "Alice",
            "age": 30
        },
        "items": [
            {"id": 1, "value": "Item1"},
            {"id": 2, "value": "Item2",}
        ],
        "active": true,
    }
    """
    expected_arguments = {
        "user": {"name": "Alice", "age": 30},
        "items": [
            {"id": 1, "value": "Item1"},
            {"id": 2, "value": "Item2"},
        ],
        "active": True,
    }

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_code_content():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"filename": "script.py", "content": "def hello():\\n    print(\'Hello, world!\')"}'
    expected_arguments = {
        "filename": "script.py",
        "content": "def hello():\n    print('Hello, world!')",
    }

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_with_escaped_quotes():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = '{"text": "He said, \\"Hello, world!\\""}'
    expected_arguments = {"text": 'He said, "Hello, world!"'}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_validate_tool_input_large_json_content():
    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[],
        task=MagicMock(),
        function_calling_llm=None,
        agent=MagicMock(),
        action=MagicMock(),
    )

    tool_input = (
        '{"data": ' + json.dumps([{"id": i, "value": i * 2} for i in range(1000)]) + "}"
    )
    expected_arguments = {"data": [{"id": i, "value": i * 2} for i in range(1000)]}

    arguments = tool_usage._validate_tool_input(tool_input)
    assert arguments == expected_arguments


def test_tool_selection_error_event_direct():
    """Test tool selection error event emission directly from ToolUsage class."""
    mock_agent = MagicMock()
    mock_agent.key = "test_key"
    mock_agent.role = "test_role"
    mock_agent.verbose = False

    mock_task = MagicMock()
    mock_tools_handler = MagicMock()

    class TestTool(BaseTool):
        name: str = "Test Tool"
        description: str = "A test tool"

        def _run(self, input: dict) -> str:
            return "test result"

    test_tool = TestTool()

    tool_usage = ToolUsage(
        tools_handler=mock_tools_handler,
        tools=[test_tool],
        task=mock_task,
        function_calling_llm=None,
        agent=mock_agent,
        action=MagicMock(),
    )

    received_events = []
    first_event_received = threading.Event()
    second_event_received = threading.Event()

    @crewai_event_bus.on(ToolSelectionErrorEvent)
    def event_handler(source, event):
        received_events.append(event)
        if event.tool_name == "Non Existent Tool":
            first_event_received.set()
        elif event.tool_name == "":
            second_event_received.set()

    with pytest.raises(Exception):  # noqa: B017
        tool_usage._select_tool("Non Existent Tool")

    assert first_event_received.wait(timeout=5), "Timeout waiting for first event"
    assert len(received_events) == 1
    event = received_events[0]
    assert isinstance(event, ToolSelectionErrorEvent)
    assert event.agent_key == "test_key"
    assert event.agent_role == "test_role"
    assert event.tool_name == "Non Existent Tool"
    assert event.tool_args == {}
    assert "Tool Name: test_tool" in event.tool_class
    assert "A test tool" in event.tool_class
    assert "don't exist" in event.error

    with pytest.raises(Exception):  # noqa: B017
        tool_usage._select_tool("")

    assert second_event_received.wait(timeout=5), "Timeout waiting for second event"
    assert len(received_events) == 2
    event = received_events[1]
    assert isinstance(event, ToolSelectionErrorEvent)
    assert event.agent_key == "test_key"
    assert event.agent_role == "test_role"
    assert event.tool_name == ""
    assert event.tool_args == {}
    assert "test_tool" in event.tool_class
    assert "forgot the Action name" in event.error


def test_tool_validate_input_error_event():
    """Test tool validation input error event emission from ToolUsage class."""
    mock_agent = MagicMock()
    mock_agent.key = "test_key"
    mock_agent.role = "test_role"
    mock_agent.verbose = False
    mock_agent._original_role = "test_role"

    mock_task = MagicMock()
    mock_tools_handler = MagicMock()

    class TestTool(BaseTool):
        name: str = "Test Tool"
        description: str = "A test tool"

        def _run(self, input: dict) -> str:
            return "test result"

    test_tool = TestTool()

    tool_usage = ToolUsage(
        tools_handler=mock_tools_handler,
        tools=[test_tool],
        task=mock_task,
        function_calling_llm=None,
        agent=mock_agent,
        action=MagicMock(tool="test_tool"),
    )
    with (
        patch("json.loads", side_effect=json.JSONDecodeError("Test Error", "", 0)),
        patch("ast.literal_eval", side_effect=ValueError),
        patch("json5.loads", side_effect=json.JSONDecodeError("Test Error", "", 0)),
        patch("json_repair.repair_json", side_effect=Exception("Failed to repair")),
    ):
        received_events = []
        condition = threading.Condition()

        @crewai_event_bus.on(ToolValidateInputErrorEvent)
        def event_handler(source, event):
            with condition:
                received_events.append(event)
                condition.notify()

        invalid_input = "invalid json {[}"
        with pytest.raises(Exception):  # noqa: B017
            tool_usage._validate_tool_input(invalid_input)

        with condition:
            if not received_events:
                condition.wait(timeout=5)

        assert len(received_events) == 1, "Expected one event to be emitted"
        event = received_events[0]
        assert isinstance(event, ToolValidateInputErrorEvent)
        assert event.agent_key == "test_key"
        assert event.agent_role == "test_role"
        assert event.tool_name == "test_tool"
        assert "must be a valid dictionary" in event.error


def test_tool_usage_finished_event_with_result():
    """Test that ToolUsageFinishedEvent is emitted with correct result attributes."""
    mock_agent = MagicMock()
    mock_agent.key = "test_agent_key"
    mock_agent.role = "test_agent_role"
    mock_agent._original_role = "test_agent_role"
    mock_agent.verbose = False

    mock_task = MagicMock()
    mock_task.delegations = 0
    mock_task.name = "Test Task"
    mock_task.description = "A test task for tool usage"
    mock_task.id = "test-task-id"

    class TestTool(BaseTool):
        name: str = "Test Tool"
        description: str = "A test tool"

        def _run(self, input: dict) -> str:
            return "test result"

    test_tool = TestTool()

    mock_tool_calling = MagicMock()
    mock_tool_calling.arguments = {"arg1": "value1"}

    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[test_tool],
        task=mock_task,
        function_calling_llm=None,
        agent=mock_agent,
        action=MagicMock(),
    )

    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(ToolUsageFinishedEvent)
    def event_handler(source, event):
        received_events.append(event)
        event_received.set()

    started_at = time.time()
    result = "test output result"
    tool_usage.on_tool_use_finished(
        tool=test_tool,
        tool_calling=mock_tool_calling,
        from_cache=False,
        started_at=started_at,
        result=result,
    )

    assert event_received.wait(timeout=5), "Timeout waiting for event"
    assert len(received_events) == 1, "Expected one event to be emitted"
    event = received_events[0]
    assert isinstance(event, ToolUsageFinishedEvent)

    assert event.agent_key == "test_agent_key"
    assert event.agent_role == "test_agent_role"
    assert event.tool_name == "test_tool"
    assert event.tool_args == {"arg1": "value1"}
    assert event.tool_class == "TestTool"
    assert event.run_attempts == 1
    assert event.delegations == 0
    assert event.from_cache is False
    assert event.output == "test output result"
    assert isinstance(event.started_at, datetime.datetime)
    assert isinstance(event.finished_at, datetime.datetime)
    assert event.type == "tool_usage_finished"


def test_tool_usage_finished_event_with_cached_result():
    """Test that ToolUsageFinishedEvent is emitted with correct result attributes when using cached result."""
    mock_agent = MagicMock()
    mock_agent.key = "test_agent_key"
    mock_agent.role = "test_agent_role"
    mock_agent._original_role = "test_agent_role"
    mock_agent.verbose = False

    mock_task = MagicMock()
    mock_task.delegations = 0
    mock_task.name = "Test Task"
    mock_task.description = "A test task for tool usage"
    mock_task.id = "test-task-id"

    class TestTool(BaseTool):
        name: str = "Test Tool"
        description: str = "A test tool"

        def _run(self, input: dict) -> str:
            return "test result"

    test_tool = TestTool()

    mock_tool_calling = MagicMock()
    mock_tool_calling.arguments = {"arg1": "value1"}

    tool_usage = ToolUsage(
        tools_handler=MagicMock(),
        tools=[test_tool],
        task=mock_task,
        function_calling_llm=None,
        agent=mock_agent,
        action=MagicMock(),
    )

    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(ToolUsageFinishedEvent)
    def event_handler(source, event):
        received_events.append(event)
        event_received.set()

    started_at = time.time()
    result = "cached test output result"
    tool_usage.on_tool_use_finished(
        tool=test_tool,
        tool_calling=mock_tool_calling,
        from_cache=True,
        started_at=started_at,
        result=result,
    )

    assert event_received.wait(timeout=5), "Timeout waiting for event"
    assert len(received_events) == 1, "Expected one event to be emitted"
    event = received_events[0]
    assert isinstance(event, ToolUsageFinishedEvent)

    assert event.agent_key == "test_agent_key"
    assert event.agent_role == "test_agent_role"
    assert event.tool_name == "test_tool"
    assert event.tool_args == {"arg1": "value1"}
    assert event.tool_class == "TestTool"
    assert event.run_attempts == 1
    assert event.delegations == 0
    assert event.from_cache is True
    assert event.output == "cached test output result"
    assert isinstance(event.started_at, datetime.datetime)
    assert isinstance(event.finished_at, datetime.datetime)
    assert event.type == "tool_usage_finished"


def test_tool_error_does_not_emit_finished_event():
    from crewai.tools.tool_calling import ToolCalling

    class FailingTool(BaseTool):
        name: str = "Failing Tool"
        description: str = "A tool that always fails"

        def _run(self, **kwargs) -> str:
            raise ValueError("Intentional failure")

    failing_tool = FailingTool().to_structured_tool()

    mock_agent = MagicMock()
    mock_agent.key = "test_agent_key"
    mock_agent.role = "test_agent_role"
    mock_agent._original_role = "test_agent_role"
    mock_agent.verbose = False
    mock_agent.fingerprint = None

    mock_task = MagicMock()
    mock_task.delegations = 0
    mock_task.name = "Test Task"
    mock_task.description = "A test task"
    mock_task.id = "test-task-id"

    mock_action = MagicMock()
    mock_action.tool = "failing_tool"
    mock_action.tool_input = "{}"

    tool_usage = ToolUsage(
        tools_handler=MagicMock(cache=None, last_used_tool=None),
        tools=[failing_tool],
        task=mock_task,
        function_calling_llm=None,
        agent=mock_agent,
        action=mock_action,
    )

    started_events = []
    error_events = []
    finished_events = []
    error_received = threading.Event()

    @crewai_event_bus.on(ToolUsageStartedEvent)
    def on_started(source, event):
        if event.tool_name == "failing_tool":
            started_events.append(event)

    @crewai_event_bus.on(ToolUsageErrorEvent)
    def on_error(source, event):
        if event.tool_name == "failing_tool":
            error_events.append(event)
            error_received.set()

    @crewai_event_bus.on(ToolUsageFinishedEvent)
    def on_finished(source, event):
        if event.tool_name == "failing_tool":
            finished_events.append(event)

    tool_calling = ToolCalling(tool_name="failing_tool", arguments={})
    tool_usage.use(calling=tool_calling, tool_string="Action: failing_tool")

    assert error_received.wait(timeout=5), "Timeout waiting for error event"
    crewai_event_bus.flush()

    assert len(started_events) >= 1, "Expected at least one ToolUsageStartedEvent"
    assert len(error_events) >= 1, "Expected at least one ToolUsageErrorEvent"
    assert len(finished_events) == 0, (
        "ToolUsageFinishedEvent should NOT be emitted after ToolUsageErrorEvent"
    )