fix: address review findings on the memory drain paths

Bugbot and CodeRabbit flagged gaps in the drain coverage: the
hierarchical `manager_agent` memory pool was never drained,
`Crew.akickoff` lacked the exception-path safety net that sync
`kickoff` has, and `finalize_session_traces` emitted the deferred
session-end `FlowFinishedEvent` without draining first. Also offloads
the pre-emit drains in the flow runtime to `asyncio.to_thread` so the
blocking wait doesn't stall other coroutines sharing the event loop.

lib/crewai/src/crewai/crew.py

@@ -1261,6 +1261,9 @@ class Crew(FlowTrackable, BaseModel):
             )
             raise
         finally:
+            # Safety net for the exception path; the success path already
+            # drained in _create_crew_output before emitting completion.
+            self._drain_memory_writes()
             clear_files(self.id)
             detach(token)
             crewai_event_bus._exit_runtime_scope(runtime_scope)
@@ -1845,8 +1848,8 @@ class Crew(FlowTrackable, BaseModel):
     def _drain_memory_writes(self) -> None:
         """Block until all pending background memory saves have completed.
 
-        Covers both the crew memory and per-agent memories — agents save
-        through ``agent.memory`` when set (see
+        Covers the crew memory, per-agent memories, and the manager agent's
+        memory — agents save through ``agent.memory`` when set (see
         ``BaseAgentExecutor._save_to_memory``), so draining only
         ``self._memory`` can miss in-flight saves. Scope/slice views are
         unwrapped to their backing ``Memory`` so each pool is drained once.
@@ -1860,6 +1863,7 @@ class Crew(FlowTrackable, BaseModel):
         candidates = [
             self._memory,
             self.memory,
+            getattr(self.manager_agent, "memory", None),
             *(getattr(agent, "memory", None) for agent in self.agents),
         ]
         for mem in candidates:

lib/crewai/src/crewai/experimental/conversational_mixin.py

@@ -1190,6 +1190,13 @@ class _ConversationalMixin:
         )
         from crewai.events.types.flow_events import FlowFinishedEvent
 
+        # 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.
+        drain_memory_writes = getattr(self, "_drain_memory_writes", None)
+        if callable(drain_memory_writes):
+            drain_memory_writes()
+
         # Only emit the session-end event when a deferred flow_started is
         # actually pending. ``_deferred_flow_started_event_id`` is set only by
         # deferred kickoffs; when finalization was not deferred, each per-turn

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -1492,8 +1492,9 @@ 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.
-            self._drain_memory_writes()
+            # listener teardown/finalization. Offloaded to a thread so the
+            # blocking drain doesn't stall other coroutines on the loop.
+            await asyncio.to_thread(self._drain_memory_writes)
             future = crewai_event_bus.emit(
                 self,
                 FlowFinishedEvent(
@@ -2307,8 +2308,9 @@ 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.
-                self._drain_memory_writes()
+                # listener teardown/finalization. Offloaded to a thread so the
+                # blocking drain doesn't stall other coroutines on the loop.
+                await asyncio.to_thread(self._drain_memory_writes)
                 future = crewai_event_bus.emit(
                     self,
                     FlowFinishedEvent(