From daede84122bf2ab538196be3a36758418899f935 Mon Sep 17 00:00:00 2001
From: Lucas Gomide <lucaslg200@gmail.com>
Date: Thu, 9 Jul 2026 12:10:00 -0300
Subject: [PATCH] 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                          |  8 ++++++--
 .../src/crewai/experimental/conversational_mixin.py    |  7 +++++++
 lib/crewai/src/crewai/flow/runtime/__init__.py         | 10 ++++++----
 3 files changed, 19 insertions(+), 6 deletions(-)

diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py
index e643e562e..7d4a5107d 100644
--- a/lib/crewai/src/crewai/crew.py
+++ b/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:
diff --git a/lib/crewai/src/crewai/experimental/conversational_mixin.py b/lib/crewai/src/crewai/experimental/conversational_mixin.py
index c7ea733e0..b956e1e9b 100644
--- a/lib/crewai/src/crewai/experimental/conversational_mixin.py
+++ b/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
diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py
index 9aa88ee11..ae385bf05 100644
--- a/lib/crewai/src/crewai/flow/runtime/__init__.py
+++ b/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(