crewAI/lib/crewai/tests/utilities/test_events.py

import threading
from datetime import datetime
import os
from unittest.mock import Mock, patch

from crewai.agent import Agent
from crewai.agents.crew_agent_executor import CrewAgentExecutor
from crewai.crew import Crew
from crewai.events.event_bus import crewai_event_bus
from crewai.events.event_listener import EventListener
from crewai.events.types.agent_events import (
    AgentExecutionCompletedEvent,
    AgentExecutionErrorEvent,
    AgentExecutionStartedEvent,
)
from crewai.events.types.crew_events import (
    CrewKickoffCompletedEvent,
    CrewKickoffFailedEvent,
    CrewKickoffStartedEvent,
    CrewTestCompletedEvent,
    CrewTestResultEvent,
    CrewTestStartedEvent,
)
from crewai.events.types.flow_events import (
    FlowCreatedEvent,
    FlowFailedEvent,
    FlowFinishedEvent,
    FlowStartedEvent,
    HumanFeedbackReceivedEvent,
    HumanFeedbackRequestedEvent,
    MethodExecutionFailedEvent,
    MethodExecutionFinishedEvent,
    MethodExecutionStartedEvent,
)
from crewai.events.types.llm_events import (
    LLMCallCompletedEvent,
    LLMCallFailedEvent,
    LLMCallStartedEvent,
    LLMStreamChunkEvent,
)
from crewai.events.types.task_events import (
    TaskCompletedEvent,
    TaskFailedEvent,
    TaskStartedEvent,
)
from crewai.events.types.tool_usage_events import (
    ToolUsageErrorEvent,
    ToolUsageFinishedEvent,
)
from crewai.flow.async_feedback.types import PendingFeedbackContext
from crewai.flow.flow import Flow, listen, start
from crewai.flow.human_feedback import human_feedback
from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on
from crewai.llm import LLM
from crewai.task import Task
from crewai.tools.base_tool import BaseTool
from pydantic import BaseModel, Field
import pytest

from ..utils import wait_for_event_handlers


@pytest.fixture(scope="module")
def base_agent():
    return Agent(
        role="base_agent",
        llm="gpt-4o-mini",
        goal="Just say hi",
        backstory="You are a helpful assistant that just says hi",
    )


@pytest.fixture(scope="module")
def base_task(base_agent):
    return Task(
        description="Just say hi",
        expected_output="hi",
        agent=base_agent,
    )


@pytest.fixture
def reset_event_listener_singleton():
    """Reset EventListener singleton for clean test state."""
    original_instance = EventListener._instance
    original_initialized = (
        getattr(EventListener._instance, "_initialized", False)
        if EventListener._instance
        else False
    )

    EventListener._instance = None

    yield

    EventListener._instance = original_instance
    if original_instance and original_initialized:
        EventListener._instance._initialized = original_initialized


@pytest.mark.vcr()
def test_crew_emits_start_kickoff_event(
    base_agent, base_task, reset_event_listener_singleton
):
    received_events = []
    mock_span = Mock()

    @crewai_event_bus.on(CrewKickoffStartedEvent)
    def handle_crew_start(source, event):
        received_events.append(event)

    mock_telemetry = Mock()
    mock_telemetry.crew_execution_span = Mock(return_value=mock_span)
    mock_telemetry.end_crew = Mock(return_value=mock_span)
    mock_telemetry.set_tracer = Mock()
    mock_telemetry.task_started = Mock(return_value=mock_span)
    mock_telemetry.task_ended = Mock(return_value=mock_span)

    with patch("crewai.events.event_listener.Telemetry", return_value=mock_telemetry):
        crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
        crew.kickoff()
    wait_for_event_handlers()

    mock_telemetry.crew_execution_span.assert_called_once_with(crew, None)
    mock_telemetry.end_crew.assert_called_once_with(crew, "hi")

    assert len(received_events) == 1
    assert received_events[0].crew_name == "TestCrew"
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "crew_kickoff_started"


@pytest.mark.vcr()
def test_crew_emits_end_kickoff_event(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(CrewKickoffCompletedEvent)
    def handle_crew_end(source, event):
        received_events.append(event)
        event_received.set()

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")

    crew.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for crew kickoff completed event"
    )
    assert len(received_events) == 1
    assert received_events[0].crew_name == "TestCrew"
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "crew_kickoff_completed"


@pytest.mark.vcr()
def test_crew_emits_test_kickoff_type_event(base_agent, base_task):
    received_events = []

    @crewai_event_bus.on(CrewTestStartedEvent)
    def handle_crew_end(source, event):
        received_events.append(event)

    @crewai_event_bus.on(CrewTestCompletedEvent)
    def handle_crew_test_end(source, event):
        received_events.append(event)

    @crewai_event_bus.on(CrewTestResultEvent)
    def handle_crew_test_result(source, event):
        received_events.append(event)

    eval_llm = LLM(model="gpt-4o-mini")
    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
    crew.test(n_iterations=1, eval_llm=eval_llm)
    wait_for_event_handlers()

    assert len(received_events) == 3
    assert received_events[0].crew_name == "TestCrew"
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "crew_test_started"
    assert received_events[1].crew_name == "TestCrew"
    assert isinstance(received_events[1].timestamp, datetime)
    assert received_events[1].type == "crew_test_result"
    assert received_events[2].crew_name == "TestCrew"
    assert isinstance(received_events[2].timestamp, datetime)
    assert received_events[2].type == "crew_test_completed"


@pytest.mark.vcr()
def test_crew_emits_kickoff_failed_event(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(CrewKickoffFailedEvent)
    def handle_crew_failed(source, event):
        received_events.append(event)
        event_received.set()

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")

    with patch.object(Crew, "_execute_tasks") as mock_execute:
        error_message = "Simulated crew kickoff failure"
        mock_execute.side_effect = Exception(error_message)

        with pytest.raises(Exception):  # noqa: B017
            crew.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for failed event"
    assert len(received_events) == 1
    assert received_events[0].error == error_message
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "crew_kickoff_failed"


@pytest.mark.vcr()
def test_crew_emits_start_task_event(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(TaskStartedEvent)
    def handle_task_start(source, event):
        received_events.append(event)
        event_received.set()

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")

    crew.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for task started event"
    assert len(received_events) == 1
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "task_started"


@pytest.mark.vcr()
def test_crew_emits_end_task_event(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(TaskCompletedEvent)
    def handle_task_end(source, event):
        received_events.append(event)
        event_received.set()

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
    crew.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for task completed event"
    assert len(received_events) == 1
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "task_completed"


@pytest.mark.vcr()
def test_task_emits_failed_event_on_execution_error(base_agent, base_task):
    received_events = []
    received_sources = []
    event_received = threading.Event()

    @crewai_event_bus.on(TaskFailedEvent)
    def handle_task_failed(source, event):
        received_events.append(event)
        received_sources.append(source)
        event_received.set()

    with patch.object(
        Task,
        "_execute_core",
    ) as mock_execute:
        error_message = "Simulated task failure"
        mock_execute.side_effect = Exception(error_message)
        agent = Agent(
            role="base_agent",
            goal="Just say hi",
            backstory="You are a helpful assistant that just says hi",
        )
        task = Task(
            description="Just say hi",
            expected_output="hi",
            agent=agent,
        )

        with pytest.raises(Exception):  # noqa: B017
            agent.execute_task(task=task)

            assert event_received.wait(timeout=5), (
                "Timeout waiting for task failed event"
            )
            assert len(received_events) == 1
            assert received_sources[0] == task
            assert received_events[0].error == error_message
            assert isinstance(received_events[0].timestamp, datetime)
            assert received_events[0].type == "task_failed"


@pytest.mark.vcr()
def test_agent_emits_execution_started_and_completed_events(base_agent, base_task):
    started_events: list[AgentExecutionStartedEvent] = []
    completed_events: list[AgentExecutionCompletedEvent] = []
    condition = threading.Condition()

    @crewai_event_bus.on(AgentExecutionStartedEvent)
    def handle_agent_start(source, event):
        with condition:
            started_events.append(event)
            condition.notify()

    @crewai_event_bus.on(AgentExecutionCompletedEvent)
    def handle_agent_completed(source, event):
        with condition:
            completed_events.append(event)
            condition.notify()

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
    crew.kickoff()

    with condition:
        success = condition.wait_for(
            lambda: len(started_events) >= 1 and len(completed_events) >= 1,
            timeout=10,
        )
    assert success, "Timeout waiting for agent execution events"

    assert len(started_events) == 1
    assert len(completed_events) == 1
    assert started_events[0].agent == base_agent
    assert started_events[0].task == base_task
    assert started_events[0].tools == []
    assert isinstance(started_events[0].task_prompt, str)
    assert (
        started_events[0].task_prompt
        == "Just say hi\n\nThis is the expected criteria for your final answer: hi\nyou MUST return the actual complete content as the final answer, not a summary."
    )
    assert isinstance(started_events[0].timestamp, datetime)
    assert started_events[0].type == "agent_execution_started"
    assert isinstance(completed_events[0].timestamp, datetime)
    assert completed_events[0].type == "agent_execution_completed"


@pytest.mark.vcr()
def test_agent_emits_execution_error_event(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(AgentExecutionErrorEvent)
    def handle_agent_start(source, event):
        received_events.append(event)
        event_received.set()

    from crewai.experimental.agent_executor import AgentExecutor

    error_message = "Error happening while sending prompt to model."
    base_agent.max_retry_limit = 0

    # Patch at the class level since agent_executor is created lazily
    with patch.object(
        AgentExecutor, "invoke", side_effect=Exception(error_message)
    ):
        with pytest.raises(Exception):  # noqa: B017
            base_agent.execute_task(
                task=base_task,
            )

        assert event_received.wait(timeout=5), (
            "Timeout waiting for agent execution error event"
        )
        assert len(received_events) == 1
        assert received_events[0].agent == base_agent
        assert received_events[0].task == base_task
        assert received_events[0].error == error_message
        assert isinstance(received_events[0].timestamp, datetime)
        assert received_events[0].type == "agent_execution_error"


class SayHiTool(BaseTool):
    name: str = Field(default="say_hi", description="The name of the tool")
    description: str = Field(
        default="Say hi", description="The description of the tool"
    )

    def _run(self) -> str:
        return "hi"


@pytest.mark.vcr()
def test_tools_emits_finished_events():
    received_events = []
    event_received = threading.Event()

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

    agent = Agent(
        role="base_agent",
        goal="Just say hi",
        backstory="You are a helpful assistant that just says hi",
        tools=[SayHiTool()],
    )

    task = Task(
        description="Just say hi",
        expected_output="hi",
        agent=agent,
    )
    crew = Crew(agents=[agent], tasks=[task], name="TestCrew")
    crew.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for tool usage finished event"
    )
    assert len(received_events) == 1
    assert received_events[0].agent_key == agent.key
    assert received_events[0].agent_role == agent.role
    assert received_events[0].tool_name == SayHiTool().name
    assert received_events[0].tool_args == "{}" or received_events[0].tool_args == {}
    assert received_events[0].type == "tool_usage_finished"
    assert isinstance(received_events[0].timestamp, datetime)


@pytest.mark.vcr()
def test_tools_emits_error_events():
    received_events = []
    lock = threading.Lock()
    all_events_received = threading.Event()

    @crewai_event_bus.on(ToolUsageErrorEvent)
    def handle_tool_end(source, event):
        with lock:
            received_events.append(event)
            # Set event when we receive at least 1 error event
            if len(received_events) >= 1:
                all_events_received.set()

    class ErrorTool(BaseTool):
        name: str = Field(
            default="error_tool", description="A tool that raises an error"
        )
        description: str = Field(
            default="This tool always raises an error",
            description="The description of the tool",
        )

        def _run(self) -> str:
            raise Exception("Simulated tool error")

    agent = Agent(
        role="base_agent",
        goal="Try to use the error tool",
        backstory="You are an assistant that tests error handling",
        tools=[ErrorTool()],
        llm=LLM(model="gpt-4o-mini"),
    )

    task = Task(
        description="Use the error tool",
        expected_output="This should error",
        agent=agent,
    )

    crew = Crew(agents=[agent], tasks=[task], name="TestCrew")
    crew.kickoff()

    assert all_events_received.wait(timeout=10), (
        "Timeout waiting for tool usage error events"
    )
    # At least one error event should be received (number varies by execution path)
    assert len(received_events) >= 1
    assert received_events[0].agent_key == agent.key
    assert received_events[0].agent_role == agent.role
    assert received_events[0].tool_name == "error_tool"
    assert received_events[0].tool_args == "{}" or received_events[0].tool_args == {}
    assert str(received_events[0].error) == "Simulated tool error"
    assert received_events[0].type == "tool_usage_error"
    assert isinstance(received_events[0].timestamp, datetime)


def test_flow_emits_start_event(reset_event_listener_singleton):
    received_events = []
    event_received = threading.Event()
    mock_span = Mock()

    @crewai_event_bus.on(FlowStartedEvent)
    def handle_flow_start(source, event):
        received_events.append(event)
        event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            return "started"

    mock_telemetry = Mock()
    mock_telemetry.flow_execution_span = Mock(return_value=mock_span)
    mock_telemetry.flow_creation_span = Mock()
    mock_telemetry.set_tracer = Mock()

    with patch("crewai.events.event_listener.Telemetry", return_value=mock_telemetry):
        # Force creation of EventListener singleton with mocked telemetry
        _ = EventListener()

        flow = TestFlow()
        flow.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for flow started event"
    mock_telemetry.flow_execution_span.assert_called_once_with(
        "TestFlow", ["begin"], "user", False
    )
    assert len(received_events) == 1
    assert received_events[0].flow_name == "TestFlow"
    assert received_events[0].type == "flow_started"


def test_flow_name_emitted_to_event_bus():
    received_events = []
    event_received = threading.Event()

    class MyFlowClass(Flow):
        name = "PRODUCTION_FLOW"

        @start()
        def start(self):
            return "Hello, world!"

    @crewai_event_bus.on(FlowStartedEvent)
    def handle_flow_start(source, event):
        received_events.append(event)
        event_received.set()

    flow = MyFlowClass()
    flow.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for flow started event"
    assert len(received_events) == 1
    assert received_events[0].flow_name == "PRODUCTION_FLOW"


def test_flow_emits_finish_event():
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(FlowFinishedEvent)
    def handle_flow_finish(source, event):
        received_events.append(event)
        event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            return "completed"

    flow = TestFlow()
    result = flow.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for finish event"
    assert len(received_events) == 1
    assert received_events[0].flow_name == "TestFlow"
    assert received_events[0].type == "flow_finished"
    assert received_events[0].result == "completed"
    assert result == "completed"


def test_flow_emits_failed_event_paired_with_started_event():
    started: list[FlowStartedEvent] = []
    failed: list[FlowFailedEvent] = []

    class BoomFlow(Flow[dict]):
        @start()
        def begin(self):
            raise RuntimeError("boom")

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(FlowStartedEvent)
        def handle_flow_started(source, event):
            started.append(event)

        @crewai_event_bus.on(FlowFailedEvent)
        def handle_flow_failed(source, event):
            failed.append(event)

        with pytest.raises(RuntimeError, match="boom"):
            BoomFlow().kickoff()
        wait_for_event_handlers()

    assert len(failed) == 1
    assert failed[0].type == "flow_failed"
    assert failed[0].flow_name == "BoomFlow"
    assert isinstance(failed[0].error, RuntimeError)
    assert str(failed[0].error) == "boom"
    assert failed[0].started_event_id == started[0].event_id


def test_suppressed_flow_failure_matches_finished_event_emission():
    finished: list[FlowFinishedEvent] = []
    failed: list[FlowFailedEvent] = []

    class SuppressedFlow(Flow):
        suppress_flow_events: bool = True

        @start()
        def begin(self):
            return "ok"

    class SuppressedBoomFlow(Flow):
        suppress_flow_events: bool = True

        @start()
        def begin(self):
            raise RuntimeError("boom")

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(FlowFinishedEvent)
        def handle_flow_finished(source, event):
            finished.append(event)

        @crewai_event_bus.on(FlowFailedEvent)
        def handle_flow_failed(source, event):
            failed.append(event)

        SuppressedFlow().kickoff()
        with pytest.raises(RuntimeError, match="boom"):
            SuppressedBoomFlow().kickoff()
        wait_for_event_handlers()

    assert len(finished) == 1
    assert len(failed) == 1


def test_abort_at_execution_start_emits_started_then_failed_events():
    started: list[FlowStartedEvent] = []
    failed: list[FlowFailedEvent] = []
    finished: list[FlowFinishedEvent] = []

    class BlockedFlow(Flow):
        @start()
        def begin(self) -> str:
            return "never runs"

    clear_all()
    try:

        @on(InterceptionPoint.EXECUTION_START)
        def block(_ctx):
            raise HookAborted(reason="blocked by policy")

        with crewai_event_bus.scoped_handlers():

            @crewai_event_bus.on(FlowStartedEvent)
            def handle_flow_started(source, event):
                started.append(event)

            @crewai_event_bus.on(FlowFailedEvent)
            def handle_flow_failed(source, event):
                failed.append(event)

            @crewai_event_bus.on(FlowFinishedEvent)
            def handle_flow_finished(source, event):
                finished.append(event)

            with pytest.raises(HookAborted):
                BlockedFlow().kickoff()
            wait_for_event_handlers()
    finally:
        clear_all()

    assert len(started) == 1
    assert len(failed) == 1
    assert finished == []
    assert failed[0].flow_name == "BlockedFlow"
    assert isinstance(failed[0].error, HookAborted)
    assert failed[0].error.reason == "blocked by policy"
    assert failed[0].started_event_id == started[0].event_id


def test_resume_emits_failed_event_paired_with_resume_started_event(tmp_path):
    started: list[FlowStartedEvent] = []
    failed: list[FlowFailedEvent] = []

    class ResumeBoomFlow(Flow):
        @start()
        def begin(self) -> str:
            return "content"

        @listen(begin)
        def after_feedback(self, _feedback):
            raise RuntimeError("boom on resume")

    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
    flow_id = "resume-failure-test"
    persistence.save_pending_feedback(
        flow_uuid=flow_id,
        context=PendingFeedbackContext(
            flow_id=flow_id,
            flow_class="ResumeBoomFlow",
            method_name="begin",
            method_output="content",
            message="Review:",
        ),
        state_data={"id": flow_id},
    )

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(FlowStartedEvent)
        def handle_flow_started(source, event):
            started.append(event)

        @crewai_event_bus.on(FlowFailedEvent)
        def handle_flow_failed(source, event):
            failed.append(event)

        flow = ResumeBoomFlow.from_pending(flow_id, persistence)
        with pytest.raises(RuntimeError, match="boom on resume"):
            flow.resume("ok")
        wait_for_event_handlers()

    assert len(started) == 1
    assert len(failed) == 1
    assert failed[0].flow_name == "ResumeBoomFlow"
    assert str(failed[0].error) == "boom on resume"
    assert failed[0].started_event_id == started[0].event_id


def test_resume_pairs_resumed_method_events_with_their_own_scope(tmp_path):
    started: list[FlowStartedEvent] = []
    finished: list[FlowFinishedEvent] = []
    method_started: list[MethodExecutionStartedEvent] = []
    method_finished: list[MethodExecutionFinishedEvent] = []

    class ResumeFlow(Flow):
        @start()
        def begin(self) -> str:
            return "content"

        @listen(begin)
        def after_feedback(self, _feedback):
            return "done"

    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
    flow_id = "resume-pairing-test"
    persistence.save_pending_feedback(
        flow_uuid=flow_id,
        context=PendingFeedbackContext(
            flow_id=flow_id,
            flow_class="ResumeFlow",
            method_name="begin",
            method_output="content",
            message="Review:",
        ),
        state_data={"id": flow_id},
    )

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(FlowStartedEvent)
        def handle_flow_started(source, event):
            started.append(event)

        @crewai_event_bus.on(FlowFinishedEvent)
        def handle_flow_finished(source, event):
            finished.append(event)

        @crewai_event_bus.on(MethodExecutionStartedEvent)
        def handle_method_started(source, event):
            method_started.append(event)

        @crewai_event_bus.on(MethodExecutionFinishedEvent)
        def handle_method_finished(source, event):
            method_finished.append(event)

        ResumeFlow.from_pending(flow_id, persistence).resume("ok")
        wait_for_event_handlers()

    resumed_started = next(e for e in method_started if e.method_name == "begin")
    resumed_finished = next(e for e in method_finished if e.method_name == "begin")

    assert resumed_finished.started_event_id == resumed_started.event_id
    assert finished[0].started_event_id == started[0].event_id


def test_resume_failing_before_method_finishes_keeps_flow_pairing(tmp_path):
    started: list[FlowStartedEvent] = []
    failed: list[FlowFailedEvent] = []
    method_failed: list[MethodExecutionFailedEvent] = []

    class ResumeFlow(Flow):
        @start()
        def begin(self) -> str:
            return "content"

        @listen(begin)
        def after_feedback(self, _feedback):
            return "done"

    persistence = SQLiteFlowPersistence(str(tmp_path / "flow.db"))
    flow_id = "resume-finalize-failure-test"
    persistence.save_pending_feedback(
        flow_uuid=flow_id,
        context=PendingFeedbackContext(
            flow_id=flow_id,
            flow_class="ResumeFlow",
            method_name="begin",
            method_output="content",
            message="Review:",
        ),
        state_data={"id": flow_id},
    )

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(FlowStartedEvent)
        def handle_flow_started(source, event):
            started.append(event)

        @crewai_event_bus.on(FlowFailedEvent)
        def handle_flow_failed(source, event):
            failed.append(event)

        @crewai_event_bus.on(MethodExecutionFailedEvent)
        def handle_method_failed(source, event):
            method_failed.append(event)

        flow = ResumeFlow.from_pending(flow_id, persistence)
        with patch.object(
            Flow,
            "_finalize_human_feedback",
            side_effect=RuntimeError("feedback collapse failed"),
        ):
            with pytest.raises(RuntimeError, match="feedback collapse failed"):
                flow.resume("ok")
        wait_for_event_handlers()

    assert len(method_failed) == 1
    assert method_failed[0].method_name == "begin"
    assert len(failed) == 1
    assert failed[0].started_event_id == started[0].event_id


def test_flow_emits_method_execution_started_event():
    received_events = []
    lock = threading.Lock()
    second_event_received = threading.Event()

    @crewai_event_bus.on(MethodExecutionStartedEvent)
    async def handle_method_start(source, event):
        with lock:
            received_events.append(event)
            if event.method_name == "second_method":
                second_event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            return "started"

        @listen("begin")
        def second_method(self):
            return "executed"

    flow = TestFlow()
    flow.kickoff()

    assert second_event_received.wait(timeout=5), (
        "Timeout waiting for second_method event"
    )
    assert len(received_events) == 2

    # Events may arrive in any order due to async handlers, so check both are present
    method_names = {event.method_name for event in received_events}
    assert method_names == {"begin", "second_method"}

    for event in received_events:
        assert event.flow_name == "TestFlow"
        assert event.type == "method_execution_started"


@pytest.mark.vcr()
def test_register_handler_adds_new_handler(base_agent, base_task):
    received_events = []
    event_received = threading.Event()

    def custom_handler(source, event):
        received_events.append(event)
        event_received.set()

    crewai_event_bus.register_handler(CrewKickoffStartedEvent, custom_handler)

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
    crew.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for handler event"
    assert len(received_events) == 1
    assert isinstance(received_events[0].timestamp, datetime)
    assert received_events[0].type == "crew_kickoff_started"


@pytest.mark.vcr()
def test_multiple_handlers_for_same_event(base_agent, base_task):
    received_events_1 = []
    received_events_2 = []
    event_received = threading.Event()

    def handler_1(source, event):
        received_events_1.append(event)

    def handler_2(source, event):
        received_events_2.append(event)
        event_received.set()

    crewai_event_bus.register_handler(CrewKickoffStartedEvent, handler_1)
    crewai_event_bus.register_handler(CrewKickoffStartedEvent, handler_2)

    crew = Crew(agents=[base_agent], tasks=[base_task], name="TestCrew")
    crew.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for handler events"
    assert len(received_events_1) == 1
    assert len(received_events_2) == 1
    assert received_events_1[0].type == "crew_kickoff_started"
    assert received_events_2[0].type == "crew_kickoff_started"


def test_flow_emits_created_event():
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(FlowCreatedEvent)
    def handle_flow_created(source, event):
        received_events.append(event)
        event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            return "started"

    flow = TestFlow()
    flow.kickoff()

    assert event_received.wait(timeout=5), "Timeout waiting for flow created event"
    assert len(received_events) == 1
    assert received_events[0].flow_name == "TestFlow"
    assert received_events[0].type == "flow_created"


def test_flow_emits_method_execution_failed_event():
    received_events = []
    event_received = threading.Event()
    error = Exception("Simulated method failure")

    @crewai_event_bus.on(MethodExecutionFailedEvent)
    def handle_method_failed(source, event):
        received_events.append(event)
        event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            raise error

    flow = TestFlow()
    with pytest.raises(Exception):  # noqa: B017
        flow.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for method execution failed event"
    )
    assert len(received_events) == 1
    assert received_events[0].method_name == "begin"
    assert received_events[0].flow_name == "TestFlow"
    assert received_events[0].type == "method_execution_failed"
    assert received_events[0].error == error


def test_flow_method_execution_started_includes_unstructured_state():
    """Test that MethodExecutionStartedEvent includes unstructured (dict) state."""
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(MethodExecutionStartedEvent)
    def handle_method_started(source, event):
        received_events.append(event)
        if event.method_name == "process":
            event_received.set()

    class TestFlow(Flow[dict]):
        @start()
        def begin(self):
            self.state["counter"] = 1
            self.state["message"] = "test"
            return "started"

        @listen("begin")
        def process(self):
            self.state["counter"] = 2
            return "processed"

    flow = TestFlow()
    flow.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for method execution started event"
    )

    begin_event = next(e for e in received_events if e.method_name == "begin")
    process_event = next(e for e in received_events if e.method_name == "process")

    assert begin_event.state is not None
    assert isinstance(begin_event.state, dict)
    assert "id" in begin_event.state

    assert process_event.state is not None
    assert isinstance(process_event.state, dict)
    assert process_event.state["counter"] == 1
    assert process_event.state["message"] == "test"


def test_flow_method_execution_started_includes_structured_state():
    """Test that MethodExecutionStartedEvent includes structured (BaseModel) state and serializes it properly."""
    received_events = []
    event_received = threading.Event()

    class FlowState(BaseModel):
        counter: int = 0
        message: str = ""
        items: list[str] = []

    @crewai_event_bus.on(MethodExecutionStartedEvent)
    def handle_method_started(source, event):
        received_events.append(event)
        if event.method_name == "process":
            event_received.set()

    class TestFlow(Flow[FlowState]):
        @start()
        def begin(self):
            self.state.counter = 1
            self.state.message = "initial"
            self.state.items = ["a", "b"]
            return "started"

        @listen("begin")
        def process(self):
            self.state.counter += 1
            return "processed"

    flow = TestFlow()
    flow.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for method execution started event"
    )

    begin_event = next(e for e in received_events if e.method_name == "begin")
    process_event = next(e for e in received_events if e.method_name == "process")

    assert begin_event.state is not None
    assert isinstance(begin_event.state, dict)
    assert begin_event.state["counter"] == 0
    assert begin_event.state["message"] == ""
    assert begin_event.state["items"] == []

    assert process_event.state is not None
    assert isinstance(process_event.state, dict)
    assert process_event.state["counter"] == 1
    assert process_event.state["message"] == "initial"
    assert process_event.state["items"] == ["a", "b"]


def test_flow_method_execution_finished_includes_serialized_state():
    """Test that MethodExecutionFinishedEvent includes properly serialized state."""
    received_events = []
    event_received = threading.Event()

    class FlowState(BaseModel):
        result: str = ""
        completed: bool = False

    @crewai_event_bus.on(MethodExecutionFinishedEvent)
    def handle_method_finished(source, event):
        received_events.append(event)
        if event.method_name == "process":
            event_received.set()

    class TestFlow(Flow[FlowState]):
        @start()
        def begin(self):
            self.state.result = "begin done"
            return "started"

        @listen("begin")
        def process(self):
            self.state.result = "process done"
            self.state.completed = True
            return "final_result"

    flow = TestFlow()
    final_output = flow.kickoff()

    assert event_received.wait(timeout=5), (
        "Timeout waiting for method execution finished event"
    )

    begin_finished = next(e for e in received_events if e.method_name == "begin")
    process_finished = next(e for e in received_events if e.method_name == "process")

    assert begin_finished.state is not None
    assert isinstance(begin_finished.state, dict)
    assert begin_finished.state["result"] == "begin done"
    assert begin_finished.state["completed"] is False
    assert begin_finished.result == "started"

    assert process_finished.state is not None
    assert isinstance(process_finished.state, dict)
    assert process_finished.state["result"] == "process done"
    assert process_finished.state["completed"] is True
    assert process_finished.result == "final_result"
    assert final_output == "final_result"


def test_suppress_flow_events_silences_method_lifecycle_events():
    """``suppress_flow_events=True`` emits no MethodExecution* events on the
    bus (used by infrastructure flows like AgentExecutor so their control-flow
    methods don't pollute traces), while default flows still emit them."""
    captured: list[tuple[str, str]] = []

    class SuppressedFlow(Flow):
        suppress_flow_events: bool = True

        @start()
        def begin(self):
            return "started"

        @listen("begin")
        def process(self):
            return "done"

    class ControlFlow(Flow):
        @start()
        def begin(self):
            return "started"

        @listen("begin")
        def process(self):
            return "done"

    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(MethodExecutionStartedEvent)
        def _on_started(source, event):
            captured.append(("started", type(source).__name__))

        @crewai_event_bus.on(MethodExecutionFinishedEvent)
        def _on_finished(source, event):
            captured.append(("finished", type(source).__name__))

        SuppressedFlow().kickoff()
        wait_for_event_handlers()
        assert [e for e in captured if e[1] == "SuppressedFlow"] == [], (
            "suppress_flow_events=True must emit no MethodExecution* events"
        )

        captured.clear()
        ControlFlow().kickoff()
        wait_for_event_handlers()
        control = [e for e in captured if e[1] == "ControlFlow"]
        assert ("started", "ControlFlow") in control
        assert ("finished", "ControlFlow") in control


def test_infrastructure_flows_suppress_flow_events_by_default():
    """Pin the infra flows that must stay silent in traces.

    The gating in ``_execute_method`` only helps if these flows actually set
    ``suppress_flow_events=True``; without this guard, removing the flag from
    AgentExecutor would silently bring back the verbose per-method trace spans.
    """
    from crewai.experimental.agent_executor import AgentExecutor
    from crewai.memory.encoding_flow import EncodingFlow
    from crewai.memory.recall_flow import RecallFlow

    assert AgentExecutor.model_fields["suppress_flow_events"].default is True

    for flow_cls in (EncodingFlow, RecallFlow):
        flow = flow_cls(storage=None, llm=None, embedder=None)
        assert flow.suppress_flow_events is True


@pytest.mark.vcr()
def test_llm_emits_call_started_event():
    started_events: list[LLMCallStartedEvent] = []
    completed_events: list[LLMCallCompletedEvent] = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def handle_llm_call_started(source, event):
        with condition:
            started_events.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def handle_llm_call_completed(source, event):
        with condition:
            completed_events.append(event)
            condition.notify()

    llm = LLM(model="gpt-4o-mini")
    llm.call("Hello, how are you?")

    with condition:
        success = condition.wait_for(
            lambda: len(started_events) >= 1 and len(completed_events) >= 1,
            timeout=10,
        )
    assert success, "Timeout waiting for LLM events"

    assert started_events[0].type == "llm_call_started"
    assert completed_events[0].type == "llm_call_completed"

    assert started_events[0].task_name is None
    assert started_events[0].agent_role is None
    assert started_events[0].agent_id is None
    assert started_events[0].task_id is None


@pytest.mark.vcr()
def test_llm_completed_event_includes_usage():
    completed_events: list[LLMCallCompletedEvent] = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def handle_llm_call_completed(source, event):
        with condition:
            completed_events.append(event)
            condition.notify()

    llm = LLM(model="gpt-4o-mini")
    llm.call("Say hello")

    with condition:
        success = condition.wait_for(
            lambda: len(completed_events) >= 1,
            timeout=10,
        )
    assert success, "Timeout waiting for LLMCallCompletedEvent"

    event = completed_events[0]
    assert event.usage is not None
    assert isinstance(event.usage, dict)
    assert event.usage.get("prompt_tokens", 0) > 0
    assert event.usage.get("completion_tokens", 0) > 0
    assert event.usage.get("total_tokens", 0) > 0


def test_llm_emits_call_failed_event():
    received_events = []
    event_received = threading.Event()

    @crewai_event_bus.on(LLMCallFailedEvent)
    def handle_llm_call_failed(source, event):
        received_events.append(event)
        event_received.set()

    error_message = "OpenAI API call failed: Simulated API failure"

    llm = LLM(model="gpt-4o-mini")
    with patch.object(
        llm, "_handle_completion", side_effect=Exception("Simulated API failure")
    ):
        with pytest.raises(Exception) as exc_info:
            llm.call("Hello, how are you?")

        assert str(exc_info.value) == "Simulated API failure"
        assert event_received.wait(timeout=5), "Timeout waiting for failed event"
        assert len(received_events) == 1
        assert received_events[0].type == "llm_call_failed"
        assert received_events[0].error == error_message
        assert received_events[0].task_name is None
        assert received_events[0].agent_role is None
        assert received_events[0].agent_id is None
        assert received_events[0].task_id is None


@pytest.mark.vcr()
def test_llm_emits_stream_chunk_events():
    """Test that LLM emits stream chunk events when streaming is enabled."""
    received_chunks = []
    event_received = threading.Event()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_stream_chunk(source, event):
        received_chunks.append(event.chunk)
        if len(received_chunks) >= 1:
            event_received.set()

    llm = LLM(model="gpt-4o", stream=True)

    response = llm.call("Tell me a short joke")

    assert event_received.wait(timeout=5), "Timeout waiting for stream chunks"

    assert len(received_chunks) > 0

    assert "".join(received_chunks) == response


@pytest.mark.vcr()
def test_llm_no_stream_chunks_when_streaming_disabled():
    """Test that LLM doesn't emit stream chunk events when streaming is disabled."""
    received_chunks = []

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_stream_chunk(source, event):
        received_chunks.append(event.chunk)

    llm = LLM(model="gpt-4o", stream=False)

    response = llm.call("Tell me a short joke")

    assert len(received_chunks) == 0

    assert response and isinstance(response, str)


@pytest.mark.vcr()
def test_streaming_fallback_to_non_streaming():
    """Test that streaming falls back to non-streaming when there's an error."""
    received_chunks = []
    fallback_called = False
    event_received = threading.Event()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_stream_chunk(source, event):
        received_chunks.append(event.chunk)
        if len(received_chunks) >= 2:
            event_received.set()

    llm = LLM(model="gpt-4o", stream=True)

    original_call = llm.call

    def mock_call(messages, tools=None, callbacks=None, available_functions=None):
        nonlocal fallback_called
        # Emit a couple of chunks to simulate partial streaming
        crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="Test chunk 1", response_id="Id", call_id="test-call-id"))
        crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="Test chunk 2", response_id="Id", call_id="test-call-id"))

        # Mark that fallback would be called
        fallback_called = True

        # Return a response as if fallback succeeded
        return "Fallback response after streaming error"

    llm.call = mock_call

    try:
        response = llm.call("Tell me a short joke")
        wait_for_event_handlers()

        assert event_received.wait(timeout=5), "Timeout waiting for stream chunks"

        assert len(received_chunks) == 2
        assert received_chunks[0] == "Test chunk 1"
        assert received_chunks[1] == "Test chunk 2"

        # Verify fallback was triggered
        assert fallback_called

        # Verify we got the fallback response
        assert response == "Fallback response after streaming error"

    finally:
        llm.call = original_call


@pytest.mark.vcr()
def test_streaming_empty_response_handling():
    """Test that streaming handles empty responses correctly."""
    received_chunks = []
    event_received = threading.Event()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_stream_chunk(source, event):
        received_chunks.append(event.chunk)
        if len(received_chunks) >= 3:
            event_received.set()

    llm = LLM(model="gpt-3.5-turbo", stream=True)

    original_call = llm.call

    # Create a mock call method that simulates empty chunks
    def mock_call(messages, tools=None, callbacks=None, available_functions=None):
        for _ in range(3):
            crewai_event_bus.emit(llm, event=LLMStreamChunkEvent(chunk="", response_id="id", call_id="test-call-id"))

        return "I apologize, but I couldn't generate a proper response. Please try again or rephrase your request."

    llm.call = mock_call

    try:
        response = llm.call("Tell me a short joke")

        assert event_received.wait(timeout=5), "Timeout waiting for empty chunks"

        assert len(received_chunks) == 3
        assert all(chunk == "" for chunk in received_chunks)

        assert "I apologize" in response and "couldn't generate" in response

    finally:
        llm.call = original_call


@pytest.mark.vcr()
def test_stream_llm_emits_event_with_task_and_agent_info():
    completed_event = []
    failed_event = []
    started_event = []
    stream_event = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallFailedEvent)
    def handle_llm_failed(source, event):
        with condition:
            failed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def handle_llm_started(source, event):
        with condition:
            started_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def handle_llm_completed(source, event):
        with condition:
            completed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_llm_stream_chunk(source, event):
        with condition:
            stream_event.append(event)
            condition.notify()

    agent = Agent(
        role="TestAgent",
        llm=LLM(model="gpt-4o-mini", stream=True),
        goal="Just say hi",
        backstory="You are a helpful assistant that just says hi",
    )
    task = Task(
        description="Just say hi",
        expected_output="hi",
        llm=LLM(model="gpt-4o-mini", stream=True),
        agent=agent,
    )

    crew = Crew(agents=[agent], tasks=[task])
    crew.kickoff()

    with condition:
        success = condition.wait_for(
            lambda: len(completed_event) >= 1
            and len(started_event) >= 1
            and len(stream_event) >= 12,
            timeout=10,
        )
    assert success, "Timeout waiting for LLM events"
    assert len(completed_event) == 1
    assert len(failed_event) == 0
    assert len(started_event) == 1
    assert len(stream_event) == 12

    all_events = completed_event + failed_event + started_event + stream_event
    all_agent_roles = [event.agent_role for event in all_events]
    all_agent_id = [event.agent_id for event in all_events]
    all_task_id = [event.task_id for event in all_events]
    all_task_name = [event.task_name for event in all_events]

    # ensure all events have the agent + task props set
    assert len(all_agent_roles) == 14
    assert len(all_agent_id) == 14
    assert len(all_task_id) == 14
    assert len(all_task_name) == 14

    assert set(all_agent_roles) == {agent.role}
    assert set(all_agent_id) == {str(agent.id)}
    assert set(all_task_id) == {str(task.id)}
    assert set(all_task_name) == {task.name or task.description}


@pytest.mark.vcr()
def test_llm_emits_event_with_task_and_agent_info(base_agent, base_task):
    completed_event: list[LLMCallCompletedEvent] = []
    failed_event: list[LLMCallFailedEvent] = []
    started_event: list[LLMCallStartedEvent] = []
    stream_event: list[LLMStreamChunkEvent] = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallFailedEvent)
    def handle_llm_failed(source, event):
        with condition:
            failed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def handle_llm_started(source, event):
        with condition:
            started_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def handle_llm_completed(source, event):
        with condition:
            completed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_llm_stream_chunk(source, event):
        with condition:
            stream_event.append(event)
            condition.notify()

    crew = Crew(agents=[base_agent], tasks=[base_task])
    crew.kickoff()

    with condition:
        success = condition.wait_for(
            lambda: len(completed_event) >= 1 and len(started_event) >= 1,
            timeout=10,
        )
    assert success, "Timeout waiting for LLM events"
    assert len(completed_event) == 1
    assert len(failed_event) == 0
    assert len(started_event) == 1
    assert len(stream_event) == 0

    all_events = completed_event + failed_event + started_event + stream_event
    all_agent_roles = [event.agent_role for event in all_events]
    all_agent_id = [event.agent_id for event in all_events]
    all_task_id = [event.task_id for event in all_events]
    all_task_name = [event.task_name for event in all_events]

    # ensure all events have the agent + task props set
    assert len(all_agent_roles) == 2
    assert len(all_agent_id) == 2
    assert len(all_task_id) == 2
    assert len(all_task_name) == 2

    assert set(all_agent_roles) == {base_agent.role}
    assert set(all_agent_id) == {str(base_agent.id)}
    assert set(all_task_id) == {str(base_task.id)}
    assert set(all_task_name) == {base_task.name or base_task.description}


@pytest.mark.vcr()
def test_llm_emits_event_with_lite_agent():
    completed_event = []
    failed_event = []
    started_event = []
    stream_event = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallFailedEvent)
    def handle_llm_failed(source, event):
        with condition:
            failed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def handle_llm_started(source, event):
        with condition:
            started_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def handle_llm_completed(source, event):
        with condition:
            completed_event.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def handle_llm_stream_chunk(source, event):
        with condition:
            stream_event.append(event)
            condition.notify()

    agent = Agent(
        role="Speaker",
        llm=LLM(model="gpt-4o-mini", stream=True),
        goal="Just say hi",
        backstory="You are a helpful assistant that just says hi",
    )
    agent.kickoff(messages=[{"role": "user", "content": "say hi!"}])

    with condition:
        success = condition.wait_for(
            lambda: len(completed_event) >= 1
            and len(started_event) >= 1
            and len(stream_event) >= 1,
            timeout=10,
        )
    assert success, "Timeout waiting for all events"

    assert len(completed_event) == 1
    assert len(failed_event) == 0
    assert len(started_event) == 1
    assert len(stream_event) >= 1

    all_events = completed_event + failed_event + started_event + stream_event
    all_agent_roles = [event.agent_role for event in all_events]
    all_agent_id = [event.agent_id for event in all_events]
    all_task_id = [event.task_id for event in all_events if event.task_id]
    all_task_name = [event.task_name for event in all_events if event.task_name]

    # ensure all events have the agent + task props set
    expected_total = 1 + 1 + len(stream_event)  # completed + started + stream
    assert len(all_agent_roles) == expected_total
    assert len(all_agent_id) == expected_total
    assert len(all_task_id) == 0
    assert len(all_task_name) == 0

    assert set(all_agent_roles) == {agent.role}
    assert set(all_agent_id) == {str(agent.id)}




@pytest.mark.vcr()
def test_llm_call_events_share_call_id():
    """All events from a single LLM call should share the same call_id."""
    import uuid

    events = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def on_start(source, event):
        with condition:
            events.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def on_complete(source, event):
        with condition:
            events.append(event)
            condition.notify()

    llm = LLM(model="gpt-4o-mini")
    llm.call("Say hi")

    with condition:
        success = condition.wait_for(lambda: len(events) >= 2, timeout=10)
    assert success, "Timeout waiting for LLM events"

    # Behavior: all events from the call share the same call_id
    assert len(events) == 2
    assert events[0].call_id == events[1].call_id
    # call_id should be a valid UUID
    uuid.UUID(events[0].call_id)


@pytest.mark.vcr()
def test_streaming_chunks_share_call_id_with_call():
    """Streaming chunks should share call_id with started/completed events."""
    events = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def on_start(source, event):
        with condition:
            events.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMStreamChunkEvent)
    def on_chunk(source, event):
        with condition:
            events.append(event)
            condition.notify()

    @crewai_event_bus.on(LLMCallCompletedEvent)
    def on_complete(source, event):
        with condition:
            events.append(event)
            condition.notify()

    llm = LLM(model="gpt-4o-mini", stream=True)
    llm.call("Say hi")

    with condition:
        success = condition.wait_for(lambda: len(events) >= 3, timeout=10)
    assert success, "Timeout waiting for streaming events"

    # Behavior: all events (started, chunks, completed) share the same call_id
    call_ids = {e.call_id for e in events}
    assert len(call_ids) == 1


@pytest.mark.vcr()
def test_separate_llm_calls_have_different_call_ids():
    """Different LLM calls should have different call_ids."""
    call_ids = []
    condition = threading.Condition()

    @crewai_event_bus.on(LLMCallStartedEvent)
    def on_start(source, event):
        with condition:
            call_ids.append(event.call_id)
            condition.notify()

    llm = LLM(model="gpt-4o-mini")
    llm.call("Say hi")
    llm.call("Say bye")

    with condition:
        success = condition.wait_for(lambda: len(call_ids) >= 2, timeout=10)
    assert success, "Timeout waiting for LLM call events"

    # Behavior: each call has its own call_id
    assert len(call_ids) == 2
    assert call_ids[0] != call_ids[1]




@patch("builtins.input", return_value="looks good")
@patch("builtins.print")
def test_human_feedback_emits_requested_and_received_events(mock_print, mock_input):
    """Test that @human_feedback decorator emits HumanFeedbackRequested and Received events."""
    requested_events = []
    received_events = []
    events_received = threading.Event()

    @crewai_event_bus.on(HumanFeedbackRequestedEvent)
    def handle_requested(source, event):
        requested_events.append(event)

    @crewai_event_bus.on(HumanFeedbackReceivedEvent)
    def handle_received(source, event):
        received_events.append(event)
        events_received.set()

    class TestFlow(Flow):
        @start()
        @human_feedback(
            message="Review:",
            emit=["approved", "rejected"],
            llm="gpt-4o-mini",
        )
        def review(self):
            return "test content"

    flow = TestFlow()

    with patch.object(flow, "_collapse_to_outcome", return_value="approved"):
        flow.kickoff()

    assert events_received.wait(timeout=5), (
        "Timeout waiting for human feedback events"
    )

    assert len(requested_events) == 1
    assert requested_events[0].type == "human_feedback_requested"
    assert requested_events[0].emit == ["approved", "rejected"]
    assert requested_events[0].message == "Review:"
    assert requested_events[0].output == "test content"

    assert len(received_events) == 1
    assert received_events[0].type == "human_feedback_received"
    assert received_events[0].feedback == "looks good"
    assert received_events[0].outcome is None

    assert flow.last_human_feedback is not None
    assert flow.last_human_feedback.outcome == "approved"


@patch("builtins.input", return_value="feedback text")
@patch("builtins.print")
def test_human_feedback_without_routing_emits_events(mock_print, mock_input):
    """Test that @human_feedback without emit still emits events."""
    requested_events = []
    received_events = []
    events_received = threading.Event()

    @crewai_event_bus.on(HumanFeedbackRequestedEvent)
    def handle_requested(source, event):
        requested_events.append(event)

    @crewai_event_bus.on(HumanFeedbackReceivedEvent)
    def handle_received(source, event):
        received_events.append(event)
        events_received.set()

    class SimpleFlow(Flow):
        @start()
        @human_feedback(message="Please review:")
        def review(self):
            return "content to review"

    flow = SimpleFlow()
    flow.kickoff()

    assert events_received.wait(timeout=5), (
        "Timeout waiting for human feedback events"
    )

    assert len(requested_events) == 1
    assert requested_events[0].emit is None

    assert len(received_events) == 1
    assert received_events[0].feedback == "feedback text"
    assert received_events[0].outcome is None


@patch("builtins.input", return_value="")
@patch("builtins.print")
def test_human_feedback_empty_feedback_emits_events(mock_print, mock_input):
    """Test that empty feedback (skipped) still emits events correctly."""
    received_events = []
    events_received = threading.Event()

    @crewai_event_bus.on(HumanFeedbackReceivedEvent)
    def handle_received(source, event):
        received_events.append(event)
        events_received.set()

    class SkipFlow(Flow):
        @start()
        @human_feedback(
            message="Review:",
            emit=["approved", "rejected"],
            llm="gpt-4o-mini",
            default_outcome="rejected",
        )
        def review(self):
            return "content"

    flow = SkipFlow()
    flow.kickoff()

    assert events_received.wait(timeout=5), (
        "Timeout waiting for human feedback events"
    )


    assert len(received_events) == 1
    assert received_events[0].feedback == ""
    assert received_events[0].outcome is None


    assert flow.last_human_feedback is not None
    assert flow.last_human_feedback.outcome == "rejected"