fix(pipecat): bound observer frame ID tracking

py/src/braintrust/integrations/pipecat/test_pipecat.py

@@ -12,11 +12,11 @@
 from braintrust import SpanCustomizer, logger, set_span_customizers
 from braintrust.integrations.pipecat import (
     BraintrustPipecatObserver,
-    PipecatIntegration,
     setup_pipecat,
     wrap_pipeline_worker,
 )
 from braintrust.integrations.test_utils import verify_autoinstrument_script
+from braintrust.integrations.versioning import detect_module_version, version_satisfies
 from braintrust.logger import Attachment
 from braintrust.test_helpers import init_test_logger
 
@@ -155,7 +155,6 @@ def _worker_runner_kwargs(**overrides):
     return kwargs
 
 
-@pytest.mark.vcr
 @pytest.mark.asyncio
 async def test_pipecat_observer_capture_audio_attachments_adds_tts_and_user_audio(memory_logger):
     TTSStartedFrame = _import("pipecat.frames.frames.TTSStartedFrame")
@@ -297,16 +296,25 @@ async def test_setup_pipecat_traces_real_pipeline_frames(memory_logger):
 
     @worker.event_handler("on_pipeline_started")
     async def on_pipeline_started(_worker, _frame):
-        await worker.queue_frames([LLMContextFrame(context), EndFrame()])
+        await worker.queue_frames([LLMContextFrame(context), EndFrame(reason="pipeline complete")])
 
     runner = WorkerRunner(**_worker_runner_kwargs())
     await runner.add_workers(worker)
     await asyncio.wait_for(runner.run(), timeout=20)
 
+    observer = next(o for o in getattr(worker, "_observer")._observers if isinstance(o, BraintrustPipecatObserver))
+    if version_satisfies(detect_module_version(importlib.import_module("pipecat"), ("pipecat",)), ">=1.12.0"):
+        assert observer.observe_every_push is False
+        assert not hasattr(observer, "_seen_frame_ids")
+    else:
+        assert observer._seen_frame_ids
+
     logs = memory_logger.pop()
     pipeline_span = _single_span(logs, "pipecat_pipeline")
     assert _span_type(pipeline_span) == "task"
     assert pipeline_span.get("metrics", {}).get("end") is not None
+    assert pipeline_span["metadata"]["terminal_frame"] == "EndFrame"
+    assert pipeline_span["metadata"]["reason"] == "pipeline complete"
 
     llm_span = _single_span(logs, "pipecat_llm_response")
     assert _span_type(llm_span) == "task"
@@ -336,7 +344,6 @@ def test_setup_and_wrap_pipeline_worker_are_idempotent():
     PipelineWorker = _import("pipecat.pipeline.worker.PipelineWorker")
     IdentityFilter = _import("pipecat.processors.filters.identity_filter.IdentityFilter")
 
-    assert PipecatIntegration.min_version == "1.3.0"
     assert setup_pipecat(project_name="test-project-pipecat-py-tracing")
     assert setup_pipecat(project_name="test-project-pipecat-py-tracing")
     assert setup_pipecat(project_name="test-project-pipecat-py-tracing", capture_audio_attachments=True)

py/src/braintrust/integrations/pipecat/tracing.py

@@ -9,6 +9,7 @@
     _pcm_to_wav,
     _resolve_audio_attachment_options,
 )
+from braintrust.integrations.versioning import detect_module_version, version_satisfies
 from braintrust.logger import NOOP_SPAN, Attachment, SpanTypeAttribute, current_span
 from braintrust.logger import start_span as _bt_start_span
 
@@ -24,8 +25,10 @@ def start_span(*args, **kwargs):
 
 
 try:
+    import pipecat
     from pipecat.observers.base_observer import BaseObserver
 except ImportError:  # pragma: no cover - exercised when Pipecat is not installed.
+    _USES_NATIVE_FRAME_DEDUPLICATION = False
 
     class BaseObserver:  # type: ignore[no-redef]
         """Fallback base so this module is importable without Pipecat."""
@@ -35,6 +38,9 @@ def __init__(self, **_kwargs: Any) -> None:
 
         async def cleanup(self) -> None:
             pass
+else:
+    _pipecat_version = detect_module_version(pipecat, ("pipecat",))
+    _USES_NATIVE_FRAME_DEDUPLICATION = version_satisfies(_pipecat_version, ">=1.12.0")
 
 
 _TERMINAL_FRAME_TYPES = {"EndFrame", "StopFrame", "CancelFrame"}
@@ -74,6 +80,9 @@ def __init__(
         trace_turns: bool = True,
         **kwargs: Any,
     ) -> None:
+        self._uses_native_frame_deduplication = _USES_NATIVE_FRAME_DEDUPLICATION
+        if self._uses_native_frame_deduplication:
+            kwargs["observe_every_push"] = False
         super().__init__(**kwargs)
         (
             self.capture_user_audio_attachments,
@@ -88,7 +97,8 @@ def __init__(
         self._parent = _current_parent_export()
         self._pipeline_span: Any | None = None
         self._pipeline_parent: str | None = None
-        self._seen_frame_ids: set[int] = set()
+        if not self._uses_native_frame_deduplication:
+            self._seen_frame_ids: set[int] = set()
         self._latest_llm_input: Any = None
         self._latest_llm_metadata: dict[str, Any] = {}
         self._llm_span: Any | None = None
@@ -111,7 +121,12 @@ async def on_pipeline_started(self) -> None:
         self._ensure_pipeline_span()
 
     async def on_process_frame(self, data: Any) -> None:
-        await self._handle_frame(getattr(data, "frame", None), processor=getattr(data, "processor", None))
+        frame = getattr(data, "frame", None)
+        processor = getattr(data, "processor", None)
+        is_terminal_at_sink = type(frame).__name__ in _TERMINAL_FRAME_TYPES and _is_pipeline_sink_processor(processor)
+        if self._uses_native_frame_deduplication and not is_terminal_at_sink:
+            return
+        await self._handle_frame(frame, processor=processor)
 
     async def on_push_frame(self, data: Any) -> None:
         await self._handle_frame(getattr(data, "frame", None), processor=getattr(data, "source", None))
@@ -128,7 +143,7 @@ async def _handle_frame(self, frame: Any, *, processor: Any = None) -> None:
             return
 
         frame_id = getattr(frame, "id", None)
-        if isinstance(frame_id, int):
+        if isinstance(frame_id, int) and not self._uses_native_frame_deduplication:
             if frame_id in self._seen_frame_ids:
                 return
             self._seen_frame_ids.add(frame_id)