[BUG] flush() returns True while sync handlers are still running, when the event type also has async handlers

Description

crewai_event_bus.flush() returns True while a sync handler is still running, if the event type it was emitted for also has an async handler registered.

emit submits the sync handlers to the thread pool but only registers the resulting future with _track_future in the sync-only branch. flush waits on _pending_futures, so for a mixed event type the sync half is invisible to it.

lib/crewai/src/crewai/events/event_bus.py:628-645 on main (ebe0082):

if sync_handlers:
    if event_type is LLMStreamChunkEvent:
        self._call_handlers(source, event, sync_handlers, state)
    else:
        ctx = contextvars.copy_context()
        sync_future = self._sync_executor.submit(
            ctx.run, self._call_handlers, source, event, sync_handlers, state
        )
        if not async_handlers:
            return self._track_future(sync_future)   # <-- only tracked here

if async_handlers:
    return self._track_future(
        asyncio.run_coroutine_threadsafe(
            self._acall_handlers(source, event, async_handlers, state),
            self._loop,
        )
    )

replay does not have this gap — it tracks the sync future unconditionally and only returns it in the sync-only case (:714-721):

if sync_handlers:
    ctx = contextvars.copy_context()
    sync_future = self._sync_executor.submit(
        ctx.run, self._call_handlers, source, event, sync_handlers, state
    )
    self._track_future(sync_future)     # <-- always tracked
    if not async_handlers:
        return sync_future

Reproduction

import asyncio, threading, time
from crewai.events.base_events import BaseEvent
from crewai.events.event_bus import crewai_event_bus


class Ev(BaseEvent):
    pass


def run(label, register_async):
    marks = []
    with crewai_event_bus.scoped_handlers():

        @crewai_event_bus.on(Ev)
        def slow_sync(source, event):
            time.sleep(1.0)
            marks.append(f"{label} SYNC FINISHED")

        if register_async:

            @crewai_event_bus.on(Ev)
            async def quick_async(source, event):
                marks.append("async finished")

        crewai_event_bus.emit("src", Ev(type=label))
        with crewai_event_bus._futures_lock:
            tracked = len(crewai_event_bus._pending_futures)

        t0 = time.monotonic()
        ok = crewai_event_bus.flush(timeout=30.0)
        print(f"{label:<24} tracked={tracked}  flush->{ok} in {time.monotonic()-t0:.2f}s  marks={marks}")
        time.sleep(1.5)
        print(f"{'':<24} 1.5s later: marks={marks}")


run("sync handlers only", register_async=False)
run("sync + async handlers", register_async=True)

Output on main:

sync handlers only       tracked=1  flush->True in 1.00s  marks=['sync handlers only SYNC FINISHED']
                         1.5s later: marks=['sync handlers only SYNC FINISHED']
sync + async handlers    tracked=0  flush->True in 0.00s  marks=['async finished']
                         1.5s later: marks=['async finished', 'sync + async handlers SYNC FINISHED']

In the mixed case flush returns True immediately with the sync handler still in flight. tracked=0 — neither future is in the pending set at that moment, because the async handler had already completed and the sync one was never added.

Impact

flush exists precisely so handlers are known to have finished before the caller moves on. lib/crewai/src/crewai/crew.py:1950-1953 states this outright:

# Ensure background memory saves finish (and emit their completed/failed events)
# before the kickoff-completed event below triggers listener teardown/finalization.
crewai_event_bus.flush()

Other call sites: conversational_mixin.py:1223, flow/runtime/__init__.py:1351 and :2498, and shutdown(wait=True) (:905), which is atexit-registered — so on interpreter exit the executor can be shut down and the process can end with sync handlers unfinished.

The triggering combination is one crewAI itself sets up. events/listeners/tracing/trace_listener.py:400 registers a sync handler on LLMCallCompletedEvent, while docs/edge/en/concepts/checkpointing.mdx:274 shows users registering an async handler on that same event. Any user who follows the docs turns every LLMCallCompletedEvent into a mixed event type, and the trace listener's writes stop being covered by flush.

Expected behaviour

flush waits for sync handlers regardless of whether async handlers are also registered for the event type — as it already does when replay emitted the event.

Additional context

The return value should stay as documented (emit's docstring at :585-591: the ThreadPoolExecutor future for sync-only, the asyncio future for async or mixed), so the fix is to track unconditionally and return the sync future only in the sync-only case — the same shape replay already uses.

Not part of this report: aemit runs only async handlers, which its docstring states explicitly, so a mixed event type emitted through aemit skips its sync handlers by design.

I have a patch and regression tests ready and will open a PR referencing this issue.

Start in lib/crewai/src/crewai/events/event_bus.py at the emit branch around lines 628-645, then compare replay around lines 714-721. Run the regression tests mentioned in the issue and verify that flush waits for sync handlers on mixed event types while emit retains its documented return behavior.