[BUG] or_() join cancels parallel listener branches that should complete

[BUG] or_() join cancels parallel listener branches that should complete

Description

When two listeners are triggered by the same method and their completion events
feed a downstream or_() listener, the runtime treats the two producer
listeners as a first-wins race. Once one producer completes, it cancels the
other producer.

or_(branch_a, branch_b) should control when the downstream listener fires. It
should not change the execution semantics of branch_a and branch_b
themselves. Both were independently triggered by their shared parent and may
contain required state updates, tool calls, or other side effects.

Steps to reproduce

import asyncio

from crewai.flow.flow import Flow, listen, or_, start


completed = []
joined = 0


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

    @listen(begin)
    async def fast_branch(self):
        await asyncio.sleep(0)
        completed.append("fast")

    @listen(begin)
    async def slow_branch(self):
        await asyncio.sleep(0.05)
        completed.append("slow")

    @listen(or_(fast_branch, slow_branch))
    def join(self):
        global joined
        joined += 1


asyncio.run(ParallelFanoutOrFlow().kickoff_async())
print(completed)
print(joined)

Actual behavior

Only the faster branch reliably completes. The slower listener is cancelled by
_execute_racing_listeners().

For synchronous listeners executed through asyncio.to_thread(), cancelling
the asyncio task cannot stop the worker thread, so the branch may instead
continue after the runtime considers it cancelled.

Expected behavior

• Both independently triggered producer listeners complete.
• The downstream or_() listener fires exactly once after the first producer
  completes.
• The OR condition does not cancel event producers.

Root cause

Flow._build_racing_groups() converts the events referenced by a downstream
OR condition back into method names. During dispatch,
_get_racing_group_for_listeners() intersects those event names with the
listener methods triggered by the current parent. If both names are present,
_execute_racing_listeners() runs them as a first-wins race and cancels the
loser.

The runtime already tracks _fired_or_listeners, which prevents the downstream
OR listener from firing more than once. Producer cancellation is unnecessary.

Proposed fix

Execute every listener selected by _find_triggered_methods() using the normal
asyncio.gather() path. Keep _fired_or_listeners responsible for deduplicating
the downstream OR listener.

Add a regression test that asserts both producer branches complete while the
OR listener fires once.

Environment

• CrewAI revision: 381fef73be1803d0b4676f88cbae57d332cb9cac
• Python: 3.13
• OS: Windows

Start with Flow._build_racing_groups(), _get_racing_group_for_listeners(), and _execute_racing_listeners(), then trace the normal asyncio.gather() dispatch path and the existing _fired_or_listeners state. Add a regression test based on the ParallelFanoutOrFlow reproduction; done means both producer branches complete and the downstream OR listener fires exactly once.