From bb15679322287ff68699baa0d1f4d6beb127b1b2 Mon Sep 17 00:00:00 2001 From: lorenzejay Date: Thu, 4 Jun 2026 15:35:44 -0700 Subject: [PATCH] refactor(tracing): rename trace event handler methods to action event handlers Updated the class to replace with for and events, improving clarity in event handling. Additionally, adjusted comments in the class to clarify the application of pending user messages in relation to state restoration and flow scope initialization. --- .../listeners/tracing/trace_listener.py | 4 ++-- lib/crewai/src/crewai/flow/runtime.py | 21 ++++++++++--------- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py index 5bcafb427..c85e6202a 100644 --- a/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py +++ b/lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py @@ -261,13 +261,13 @@ class TraceCollectionListener(BaseEventListener): def on_conversation_message_added( source: Any, event: ConversationMessageAddedEvent ) -> None: - self._handle_trace_event("conversation_message_added", source, event) + self._handle_action_event("conversation_message_added", source, event) @event_bus.on(ConversationRouteSelectedEvent) def on_conversation_route_selected( source: Any, event: ConversationRouteSelectedEvent ) -> None: - self._handle_trace_event("conversation_route_selected", source, event) + self._handle_action_event("conversation_route_selected", source, event) @event_bus.on(FlowFinishedEvent) def on_flow_finished(source: Any, event: FlowFinishedEvent) -> None: diff --git a/lib/crewai/src/crewai/flow/runtime.py b/lib/crewai/src/crewai/flow/runtime.py index 02c673f14..7dfefd3d8 100644 --- a/lib/crewai/src/crewai/flow/runtime.py +++ b/lib/crewai/src/crewai/flow/runtime.py @@ -2202,16 +2202,6 @@ class Flow(_ConversationalMixin, BaseModel, Generic[T], metaclass=FlowMeta): if filtered_inputs: self._initialize_state(filtered_inputs) - # Conversational hook: apply the pending user message AFTER state - # restore so it survives ``self.persistence.load_state(...)``. - # ``handle_turn`` stashes the message on ``self._pending_user_message`` - # before calling ``kickoff``; this drains it. - if ( - getattr(type(self), "conversational", False) - and self._pending_user_message is not None - ): - self._apply_pending_conversational_turn() - defer_trace_finalization = self._should_defer_trace_finalization() deferred_started_event_id = self._deferred_flow_started_event_id should_emit_flow_started = not ( @@ -2262,6 +2252,17 @@ class Flow(_ConversationalMixin, BaseModel, Generic[T], metaclass=FlowMeta): # with implicit "crew" execution_type. get_env_context() + # Conversational hook: apply the pending user message AFTER state + # restore and AFTER flow scope initialization, so transcript events + # are parented under the current conversation trace. + # ``handle_turn`` stashes the message on ``self._pending_user_message`` + # before calling ``kickoff``; this drains it. + if ( + getattr(type(self), "conversational", False) + and self._pending_user_message is not None + ): + self._apply_pending_conversational_turn() + if inputs is not None and "id" not in inputs: self._initialize_state(inputs)