fix(events): track the sync-handler future when async handlers also exist

[LHMQ878]

Description

emit registered the sync-handler future with _track_future only when the event type had no async handlers. flush waits on the tracked set, so for a mixed event type it returned True with the sync handlers still running.

Fixes #6745

Root cause

event_bus.py:628-645 on main:

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

Tracking was folded into the return, so it happened only on the path that returns the sync future. When async handlers also exist, control falls through to the async branch and the sync future is dropped — created, running, and unreachable from _pending_futures.

replay (:714-721) already does the right thing: it calls _track_future(sync_future) unconditionally and only returns it in the sync-only case. This PR makes emit match.

Measured on main:

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

flush returned in 0.00s and the sync handler finished 1.5s afterwards. (tracked=1 is the async future, whose done-callback had not yet fired; the sync one was never added.)

Impact

flush exists so callers know handlers finished. crew.py:1950-1953 says so:

# 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 at interpreter exit the executor could be shut down with sync handlers still in flight.

The triggering combination is the documented one: listeners/tracing/trace_listener.py:400 registers a sync handler on LLMCallCompletedEvent, and docs/edge/en/concepts/checkpointing.mdx:274 shows users registering an async handler on that same event.

The change

sync_future = self._sync_executor.submit(
    ctx.run, self._call_handlers, source, event, sync_handlers, state
)
self._track_future(sync_future)
if not async_handlers:
    return sync_future

The documented return-value contract is unchanged — emit's docstring (:585-591) promises the ThreadPoolExecutor future for sync-only handlers and the asyncio future for async or mixed. Only tracking moves; the sync future is still returned exactly when it was before.

Tests

lib/crewai/tests/utilities/events/test_flush.py, new. No existing test emitted an event that had both a sync and an async handler and then asserted on flush, which is why the gap survived.

test	asserts
test_flush_waits_for_sync_handlers_alongside_async_handlers	the regression: a mixed event type has its sync handlers awaited by flush
test_flush_waits_for_sync_handlers_when_there_are_no_async_handlers	the sync-only path, which already waited, cannot regress
test_emit_tracks_the_sync_future_when_async_handlers_exist	both futures reach _pending_futures, not just the async one
test_emit_returns_the_sync_future_only_when_there_are_no_async_handlers	the documented return-value contract still holds

Two notes for review:

• Each test asserts on the handler's own completion marker, not on elapsed time, and holds the sync handler on a threading.Event gate that is released only after flush is already blocking. A pass therefore cannot come from the handler having finished early.
• ..._tracks_the_sync_future... compares the tracked set as a delta and keeps both handlers blocked while it reads it. _pending_futures belongs to the process-wide singleton bus and _track_future's done-callback discards completed futures, so neither an absolute count nor a read after completion would be meaningful.

Control experiment

Reverting only event_bus.py and keeping the new tests:

test	on unfixed source
..._sync_handlers_alongside_async_handlers	failed
..._tracks_the_sync_future_when_async_handlers_exist	failed
..._when_there_are_no_async_handlers	passed
..._returns_the_sync_future_only_when...	passed

Verification

check	result
tests/utilities/events/test_flush.py	4 passed
tests/utilities/events/ + tests/events/	226 passed; error/failure set name-identical to main (9 pre-existing VCR-cassette errors in test_event_ordering.py, unchanged)
ruff check / ruff format --check	clean

Out of scope

aemit runs only async handlers, and its docstring says so explicitly, so a mixed event type emitted through aemit skips its sync handlers by design. It has no call sites in src/. Left alone here.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: c8d8a327-0000-44cc-93fe-6a13e451e0f2

📥 Commits

Reviewing files that changed from the base of the PR and between 5044af8 and f4c587d.

📒 Files selected for processing (1)

• lib/crewai/tests/utilities/events/test_flush.py

---

📝 Walkthrough

Walkthrough

The event bus now tracks synchronous handler futures for mixed sync/async events, allowing flush to wait for all handlers while preserving emit return behavior. Regression tests cover mixed, sync-only, and async-only handlers with deterministic synchronization.

Changes

Event flush tracking

Layer / File(s)	Summary
Track and validate handler futures lib/crewai/src/crewai/events/event_bus.py, lib/crewai/tests/utilities/events/test_flush.py	Synchronous futures are always added to the pending set, while emit continues returning the synchronous future for sync-only events and the async future for mixed events. Tests verify timeout and completion behavior across handler combinations.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 38.46% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the main fix: tracking synchronous-handler futures when asynchronous handlers are also present.
Description check	✅ Passed	The description directly explains the bug, fix, impact, tests, verification, and scope of the event-bus changes.
Linked Issues check	✅ Passed	The implementation satisfies issue #6745 by tracking sync futures for mixed events, preserving return values, and adding targeted regression tests.
Out of Scope Changes check	✅ Passed	The changes are limited to event-bus future tracking and focused regression tests; no unrelated code changes are indicated.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 2

🧹 Nitpick comments (1)

lib/crewai/tests/utilities/events/test_flush.py (1)

24-26: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Prefer the observable flush() contract over private future-set inspection.

_pending_futures and _futures_lock are implementation details. Fold this coverage into the blocked-gate flush() regression instead of asserting the internal set delta.

As per coding guidelines, unit tests for new functionality must focus on behavior rather than implementation details.

Also applies to: 85-112

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@lib/crewai/tests/utilities/events/test_flush.py` around lines 24 - 26,
Replace the private `_tracked_futures()` inspection and related
`_pending_futures`/`_futures_lock` assertions with coverage through the public
`crewai_event_bus.flush()` contract. Fold this setup and verification into the
blocked-gate flush regression, asserting observable completion or blocking
behavior rather than internal future-set changes.

Source: Coding guidelines

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@lib/crewai/tests/utilities/events/test_flush.py`:
- Around line 115-141: Strengthen
test_emit_returns_the_sync_future_only_when_there_are_no_async_handlers by
gating the sync_handler for the mixed event, asserting mixed_future resolves
before releasing the gate, then releasing the sync handler and calling flush()
for cleanup. Keep the existing sync-only assertions and ensure the test proves
emit returns the async future rather than the current mixed sync future.
- Around line 47-57: Update lib/crewai/tests/utilities/events/test_flush.py
lines 47-57 to add an async-completion event, wait for the async handler in the
mixed case, assert flush() returns False while gate remains closed, then release
gate and verify a subsequent flush succeeds; update lines 74-82 to assert
flush() returns False before releasing gate. Keep the tests focused on
observable flush behavior.

---

Nitpick comments:
In `@lib/crewai/tests/utilities/events/test_flush.py`:
- Around line 24-26: Replace the private `_tracked_futures()` inspection and
related `_pending_futures`/`_futures_lock` assertions with coverage through the
public `crewai_event_bus.flush()` contract. Fold this setup and verification
into the blocked-gate flush regression, asserting observable completion or
blocking behavior rather than internal future-set changes.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 90f5c2bc-91c9-4509-a402-4c629435106a

📥 Commits

Reviewing files that changed from the base of the PR and between ebe0082 and 5044af8.

📒 Files selected for processing (2)

• lib/crewai/src/crewai/events/event_bus.py
• lib/crewai/tests/utilities/events/test_flush.py