Prevent crew inside flow to finish trace batch prematurely on failure

Author: heitorado

lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py

@@ -221,11 +221,12 @@ def on_crew_completed(source: Any, event: CrewKickoffCompletedEvent) -> None:
         @event_bus.on(CrewKickoffFailedEvent)
         def on_crew_failed(source: Any, event: CrewKickoffFailedEvent) -> None:
             self._handle_trace_event("crew_kickoff_failed", source, event)
-            if self.first_time_handler.is_first_time:
-                self.first_time_handler.mark_events_collected()
-                self.first_time_handler.handle_execution_completion()
-            else:
-                self.batch_manager.finalize_batch()
+            if self.batch_manager.batch_owner_type == "crew":
+                if self.first_time_handler.is_first_time:
+                    self.first_time_handler.mark_events_collected()
+                    self.first_time_handler.handle_execution_completion()
+                else:
+                    self.batch_manager.finalize_batch()
 
         @event_bus.on(TaskStartedEvent)
         def on_task_started(source: Any, event: TaskStartedEvent) -> None: