From 97875a9777267208a33495c1ff57d7a36c936e08 Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Mon, 28 Sep 2026 16:11:08 -0400 Subject: [PATCH 1/4] fix(pipecat): bound observer frame ID tracking --- .../integrations/pipecat/test_pipecat.py | 13 +++++++++++++ .../braintrust/integrations/pipecat/tracing.py | 16 ++++++++++++++-- 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/py/src/braintrust/integrations/pipecat/test_pipecat.py b/py/src/braintrust/integrations/pipecat/test_pipecat.py index cc25a21e9..ed8365c20 100644 --- a/py/src/braintrust/integrations/pipecat/test_pipecat.py +++ b/py/src/braintrust/integrations/pipecat/test_pipecat.py @@ -17,6 +17,7 @@ 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 @@ -86,6 +87,18 @@ def test_pipecat_observer_filters_metrics_from_other_processors(): assert observer._llm_metrics == {"prompt_tokens": 10, "completion_tokens": 5, "tokens": 15} +def test_observer_bounds_frame_id_tracking_by_pipecat_version(): + pipecat = importlib.import_module("pipecat") + version = detect_module_version(pipecat, ("pipecat",)) + observer = BraintrustPipecatObserver() + + if version_satisfies(version, ">=1.12.0"): + assert observer.observe_every_push is False + assert not hasattr(observer, "_seen_frame_ids") + else: + assert observer._seen_frame_ids == set() + + @pytest.mark.asyncio async def test_span_customizer_redacts_incremental_tts_input(memory_logger): TTSStartedFrame = _import("pipecat.frames.frames.TTSStartedFrame") diff --git a/py/src/braintrust/integrations/pipecat/tracing.py b/py/src/braintrust/integrations/pipecat/tracing.py index d84d94466..4f9552838 100644 --- a/py/src/braintrust/integrations/pipecat/tracing.py +++ b/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,6 +121,8 @@ async def on_pipeline_started(self) -> None: self._ensure_pipeline_span() async def on_process_frame(self, data: Any) -> None: + if self._uses_native_frame_deduplication: + return await self._handle_frame(getattr(data, "frame", None), processor=getattr(data, "processor", None)) async def on_push_frame(self, data: Any) -> None: @@ -128,7 +140,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) From d59bf5436e3bc0b4c1754eaea4e8a6e58e5ffbcf Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Mon, 28 Sep 2026 16:20:37 -0400 Subject: [PATCH 2/4] fix(pipecat): close spans on terminal frames --- py/src/braintrust/integrations/pipecat/test_pipecat.py | 4 +++- py/src/braintrust/integrations/pipecat/tracing.py | 7 +++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/py/src/braintrust/integrations/pipecat/test_pipecat.py b/py/src/braintrust/integrations/pipecat/test_pipecat.py index ed8365c20..f14aa4c4f 100644 --- a/py/src/braintrust/integrations/pipecat/test_pipecat.py +++ b/py/src/braintrust/integrations/pipecat/test_pipecat.py @@ -310,7 +310,7 @@ 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) @@ -320,6 +320,8 @@ async def on_pipeline_started(_worker, _frame): 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" diff --git a/py/src/braintrust/integrations/pipecat/tracing.py b/py/src/braintrust/integrations/pipecat/tracing.py index 4f9552838..554c70048 100644 --- a/py/src/braintrust/integrations/pipecat/tracing.py +++ b/py/src/braintrust/integrations/pipecat/tracing.py @@ -121,9 +121,12 @@ async def on_pipeline_started(self) -> None: self._ensure_pipeline_span() async def on_process_frame(self, data: Any) -> None: - if self._uses_native_frame_deduplication: + 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(getattr(data, "frame", None), processor=getattr(data, "processor", None)) + 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)) From a3b3623a5a9fcfbbb0d13d6dae712803ddcb75ae Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Tue, 29 Sep 2026 14:55:02 +0000 Subject: [PATCH 3/4] test(pipecat): fold frame ID tracking check into real pipeline test Co-Authored-By: Claude Opus 5.5 --- .../integrations/pipecat/test_pipecat.py | 19 +++++++------------ 1 file changed, 7 insertions(+), 12 deletions(-) diff --git a/py/src/braintrust/integrations/pipecat/test_pipecat.py b/py/src/braintrust/integrations/pipecat/test_pipecat.py index f14aa4c4f..c67672fa2 100644 --- a/py/src/braintrust/integrations/pipecat/test_pipecat.py +++ b/py/src/braintrust/integrations/pipecat/test_pipecat.py @@ -87,18 +87,6 @@ def test_pipecat_observer_filters_metrics_from_other_processors(): assert observer._llm_metrics == {"prompt_tokens": 10, "completion_tokens": 5, "tokens": 15} -def test_observer_bounds_frame_id_tracking_by_pipecat_version(): - pipecat = importlib.import_module("pipecat") - version = detect_module_version(pipecat, ("pipecat",)) - observer = BraintrustPipecatObserver() - - if version_satisfies(version, ">=1.12.0"): - assert observer.observe_every_push is False - assert not hasattr(observer, "_seen_frame_ids") - else: - assert observer._seen_frame_ids == set() - - @pytest.mark.asyncio async def test_span_customizer_redacts_incremental_tts_input(memory_logger): TTSStartedFrame = _import("pipecat.frames.frames.TTSStartedFrame") @@ -316,6 +304,13 @@ async def on_pipeline_started(_worker, _frame): 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" From b70bfc16faa1a02c83dac9ba1da776a8aa3bd70f Mon Sep 17 00:00:00 2001 From: Abhijeet Prasad Date: Tue, 29 Sep 2026 14:56:18 +0000 Subject: [PATCH 4/4] test(pipecat): drop no-op VCR marker and constant assertion Co-Authored-By: Claude Opus 5.5 --- py/src/braintrust/integrations/pipecat/test_pipecat.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/py/src/braintrust/integrations/pipecat/test_pipecat.py b/py/src/braintrust/integrations/pipecat/test_pipecat.py index c67672fa2..ab8511857 100644 --- a/py/src/braintrust/integrations/pipecat/test_pipecat.py +++ b/py/src/braintrust/integrations/pipecat/test_pipecat.py @@ -12,7 +12,6 @@ from braintrust import SpanCustomizer, logger, set_span_customizers from braintrust.integrations.pipecat import ( BraintrustPipecatObserver, - PipecatIntegration, setup_pipecat, wrap_pipeline_worker, ) @@ -156,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") @@ -346,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)