From 67a3444bb38ea30565e266bb37e6442be20f067a Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Thu, 9 Jul 2026 12:21:06 -0300 Subject: [PATCH] fix: flush event bus after memory drain in flow completion paths Bugbot flagged that flow paths went straight from the memory drain to `FlowFinishedEvent`, while crew kickoff flushes the bus in between. Save completion events emitted during the drain could still have pending async handlers when flow-finished triggered trace teardown. Adds a `crewai_event_bus.flush()` after the drain at both flow runtime emit sites and in `finalize_session_traces`, mirroring `Crew._create_crew_output`. --- .../crewai/experimental/conversational_mixin.py | 2 ++ lib/crewai/src/crewai/flow/runtime/__init__.py | 14 ++++++++++---- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/lib/crewai/src/crewai/experimental/conversational_mixin.py b/lib/crewai/src/crewai/experimental/conversational_mixin.py index b956e1e9b..490565e9d 100644 --- a/lib/crewai/src/crewai/experimental/conversational_mixin.py +++ b/lib/crewai/src/crewai/experimental/conversational_mixin.py @@ -1193,9 +1193,11 @@ class _ConversationalMixin: # Background memory saves must finish (and emit their completed/failed # events) before the session-end flow_finished / batch finalization # below tears down listeners, mirroring the non-deferred kickoff path. + # The flush then waits for those events' async bus handlers. drain_memory_writes = getattr(self, "_drain_memory_writes", None) if callable(drain_memory_writes): drain_memory_writes() + crewai_event_bus.flush() # Only emit the session-end event when a deferred flow_started is # actually pending. ``_deferred_flow_started_event_id`` is set only by diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index ae385bf05..a6c73e697 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -1492,9 +1492,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ): # Background memory saves must finish (and emit their # completed/failed events) before flow-finished triggers - # listener teardown/finalization. Offloaded to a thread so the - # blocking drain doesn't stall other coroutines on the loop. + # listener teardown/finalization; the flush then waits for those + # events' async handlers, mirroring Crew._create_crew_output. + # Offloaded to a thread so the blocking waits don't stall other + # coroutines on the loop. await asyncio.to_thread(self._drain_memory_writes) + await asyncio.to_thread(crewai_event_bus.flush) future = crewai_event_bus.emit( self, FlowFinishedEvent( @@ -2308,9 +2311,12 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): if not self._should_defer_trace_finalization(): # Background memory saves must finish (and emit their # completed/failed events) before flow-finished triggers - # listener teardown/finalization. Offloaded to a thread so the - # blocking drain doesn't stall other coroutines on the loop. + # listener teardown/finalization; the flush then waits for + # those events' async handlers, mirroring + # Crew._create_crew_output. Offloaded to a thread so the + # blocking waits don't stall other coroutines on the loop. await asyncio.to_thread(self._drain_memory_writes) + await asyncio.to_thread(crewai_event_bus.flush) future = crewai_event_bus.emit( self, FlowFinishedEvent(