fix(pipecat): close spans on terminal frames

Author: AbhiPrasad

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"

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))