From 4d9e258e554584f65703e9e546af8165f3e5b6a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 14 May 2026 22:53:10 -0700 Subject: [PATCH] Collapse _pipeline_finished_event into BaseTask._finished_event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PipelineTask had its own ``_pipeline_finished_event`` that signalled "pipeline run has truly finished" — the same role ``BaseTask._finished_event`` plays for bus-only tasks. They were two events with the same intent. Set ``_finished_event`` directly when the pipeline-end frame propagates through the sink, drop the now-redundant field, and drop the ``clear()`` after wait so the event stays set for the lifetime of the task. As a side-effect, ``await pipeline_task.wait()`` from outside now resolves at the moment the pipeline finishes, matching the semantics of bus-only tasks. --- src/pipecat/pipeline/task.py | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 985c5527c..c9136c6af 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -396,9 +396,6 @@ class PipelineTask(BaseTask): # StopFrame) has been received at the end of the pipeline. self._pipeline_end_event = asyncio.Event() - # This event is set when the pipeline truly finishes. - self._pipeline_finished_event = asyncio.Event() - # When bridged, wrap the user pipeline with bus edge processors # so frames tee onto the bus at the source/sink and incoming bus # frames are injected back into the local pipeline. The edges @@ -803,12 +800,12 @@ class PipelineTask(BaseTask): self._pipeline_end_event.clear() - # We are really done. - self._pipeline_finished_event.set() + # We are really done. Setting ``_finished_event`` makes + # ``BaseTask.wait()`` resolve for callers awaiting this task. + self._finished_event.set() async def _wait_for_pipeline_finished(self): - await self._pipeline_finished_event.wait() - self._pipeline_finished_event.clear() + await self._finished_event.wait() # Make sure we wait for the main task to complete. if self._process_push_task: await self._process_push_task