andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-09 17:05:08 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

fix: flush event bus after memory drain in flow completion paths

Browse Source 
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`.
This commit is contained in:
Lucas Gomide
2026-07-09 12:21:06 -03:00
parent daede84122
commit 67a3444bb3
2 changed files with 12 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/crewai/src/crewai/experimental/conversational_mixin.py
View File

@@ -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

14
lib/crewai/src/crewai/flow/runtime/__init__.py
View File

@@ -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(