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(