fix(tracing): serialize concurrent trace finalization

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

@@ -1,7 +1,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from logging import getLogger
-from threading import Condition, Lock
+from threading import Condition, Lock, RLock
 import time
 from typing import Any
 import uuid
@@ -63,7 +63,7 @@ def __init__(self) -> None:
         self._pending_events_lock = Lock()
         self._pending_events_cv = Condition(self._pending_events_lock)
         self._pending_events_count = 0
-        self._finalize_lock = Lock()
+        self._finalize_lock = RLock()
 
         self.is_current_batch_ephemeral = False
         self.trace_batch_id: str | None = None
@@ -319,6 +319,11 @@ def _send_events_to_backend(self) -> int:
 
     def finalize_batch(self) -> TraceBatch | None:
         """Finalize batch and return it for sending"""
+        with self._finalize_lock:
+            return self._finalize_batch()
+
+    def _finalize_batch(self) -> TraceBatch | None:
+        """Finalize the current batch while holding ``_finalize_lock``."""
 
         if self._batch_finalized:
             return None

lib/crewai/tests/tracing/test_tracing.py

@@ -1,5 +1,5 @@
 import os
-from threading import Thread
+from threading import Barrier, Event, Lock, Thread
 from unittest.mock import MagicMock, Mock, patch
 
 import pytest
@@ -985,6 +985,83 @@ def finalize() -> None:
         assert results == [True, True]
         mock_finalize.assert_called_once()
 
+    def test_finalize_batch_sends_events_once_when_called_concurrently(self) -> None:
+        """Concurrent public finalizers must not resend an acknowledged event."""
+        batch_manager = TraceBatchManager()
+        batch_manager.current_batch = TraceBatch(
+            user_context={"privacy_level": "standard"},
+            execution_metadata={"execution_type": "crew", "crew_name": "test"},
+        )
+        batch_manager.trace_batch_id = "concurrent-finalize-batch"
+        batch_manager.backend_initialized = True
+        batch_manager.event_buffer = [
+            TraceEvent(
+                type="llm_call_started",
+                timestamp="2026-01-01T00:00:00",
+                event_id="concurrent-event",
+                emission_sequence=1,
+            )
+        ]
+        start = Barrier(2)
+        second_send_started = Event()
+        send_lock = Lock()
+        errors: list[BaseException] = []
+        send_count = 0
+
+        def send_events(*_args: object, **_kwargs: object) -> MagicMock:
+            nonlocal send_count
+            with send_lock:
+                send_count += 1
+                is_first_send = send_count == 1
+            if is_first_send:
+                # Bounded overlap window, not a synchronisation point: once the
+                # finalizers are serialized the second send never starts, so this
+                # times out and the call-count assertions decide the result.
+                second_send_started.wait(timeout=0.25)
+            else:
+                second_send_started.set()
+            return MagicMock(status_code=200)
+
+        def finalize() -> None:
+            try:
+                start.wait(timeout=5)
+                batch_manager.finalize_batch()
+            except BaseException as error:
+                errors.append(error)
+
+        with (
+            patch(
+                "crewai.events.listeners.tracing.trace_batch_manager.is_tracing_enabled_in_context",
+                return_value=True,
+            ),
+            patch(
+                "crewai.events.listeners.tracing.trace_batch_manager.should_auto_collect_first_time_traces",
+                return_value=True,
+            ),
+            patch.object(
+                batch_manager.plus_api,
+                "send_trace_events",
+                side_effect=send_events,
+            ) as send,
+            patch.object(
+                batch_manager.plus_api,
+                "finalize_trace_batch",
+                return_value=MagicMock(
+                    status_code=200, json=MagicMock(return_value={})
+                ),
+            ) as finalize_backend,
+        ):
+            threads = [Thread(target=finalize), Thread(target=finalize)]
+            for thread in threads:
+                thread.start()
+            for thread in threads:
+                thread.join(timeout=5)
+
+        assert not errors
+        assert all(not thread.is_alive() for thread in threads)
+        send.assert_called_once()
+        finalize_backend.assert_called_once()
+
     def test_ephemeral_batch_includes_anon_id(self):
         """Test that ephemeral batch initialization sends anon_id from get_user_id()"""
         fake_user_id = "abc123def456"