From df1b071a130912077c6f904773f996dcfb771f36 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 13:37:42 -0700
Subject: [PATCH 1/7] Move create_task and cancel_task from FrameProcessor to
 BaseObject

Lift the task manager wiring (`_task_manager`, `task_manager` property,
`create_task`, `cancel_task`, and `setup(task_manager)`) up to
`BaseObject`. Owners propagate the task manager to their child
`BaseObject`s via `await child.setup(task_manager)`, matching the
existing convention.

Removes duplicated `_task_manager` / `task_manager` property / setup
implementations from `FrameProcessor`, `FrameProcessorMetrics`,
`UserIdleController`, `UserTurnController`,
`BaseUserTurnStartStrategy`, and `BaseUserTurnStopStrategy`.
---
 src/pipecat/processors/frame_processor.py     | 52 +--------------
 .../metrics/frame_processor_metrics.py        | 25 -------
 src/pipecat/turns/user_idle_controller.py     | 25 +------
 .../base_user_turn_start_strategy.py          | 17 -----
 .../user_stop/base_user_turn_stop_strategy.py | 17 -----
 src/pipecat/turns/user_turn_controller.py     | 18 ++---
 src/pipecat/utils/base_object.py              | 65 +++++++++++++++++++
 7 files changed, 74 insertions(+), 145 deletions(-)

diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py
index 05abfceb4..527b19dd0 100644
--- a/src/pipecat/processors/frame_processor.py
+++ b/src/pipecat/processors/frame_processor.py
@@ -17,7 +17,7 @@ import asyncio
 import dataclasses
 import traceback
 import warnings
-from collections.abc import Awaitable, Callable, Coroutine
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from enum import Enum
 from typing import (
@@ -217,9 +217,6 @@ class FrameProcessor(BaseObject):
         # Clock
         self._clock: BaseClock | None = None
 
-        # Task Manager
-        self._task_manager: BaseTaskManager | None = None
-
         # Observer
         self._observer: BaseObserver | None = None
 
@@ -368,20 +365,6 @@ class FrameProcessor(BaseObject):
         """
         return self._report_only_initial_ttfb
 
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Get the task manager for this processor.
-
-        Returns:
-            The task manager instance.
-
-        Raises:
-            Exception: If the task manager is not initialized.
-        """
-        if not self._task_manager:
-            raise Exception(f"{self} TaskManager is still not initialized.")
-        return self._task_manager
-
     @property
     def pipeline_task(self) -> PipelineTask | None:
         """Get the :class:`PipelineTask` this processor is running in.
@@ -511,43 +494,14 @@ class FrameProcessor(BaseObject):
         await self.stop_processing_metrics()
         await self.stop_text_aggregation_metrics()
 
-    def create_task(self, coroutine: Coroutine, name: str | None = None) -> asyncio.Task:
-        """Create a new task managed by this processor.
-
-        Args:
-            coroutine: The coroutine to run in the task.
-            name: Optional name for the task.
-
-        Returns:
-            The created asyncio task.
-        """
-        if name:
-            name = f"{self}::{name}"
-        else:
-            name = f"{self}::{coroutine.cr_code.co_name}"
-        return self.task_manager.create_task(coroutine, name)
-
-    async def cancel_task(self, task: asyncio.Task, timeout: float | None = 1.0):
-        """Cancel a task managed by this processor.
-
-        A default timeout if 1 second is used in order to avoid potential
-        freezes caused by certain libraries that swallow
-        `asyncio.CancelledError`.
-
-        Args:
-            task: The task to cancel.
-            timeout: Optional timeout for task cancellation.
-        """
-        await self.task_manager.cancel_task(task, timeout)
-
     async def setup(self, setup: FrameProcessorSetup):
         """Set up the processor with required components.
 
         Args:
             setup: Configuration object containing setup parameters.
         """
+        await super().setup(setup.task_manager)
         self._clock = setup.clock
-        self._task_manager = setup.task_manager
         self._observer = setup.observer
         self._pipeline_task = setup.pipeline_task
 
@@ -555,7 +509,7 @@ class FrameProcessor(BaseObject):
         self.__create_input_task()
 
         if self._metrics is not None:
-            await self._metrics.setup(self._task_manager)
+            await self._metrics.setup(self.task_manager)
 
     async def cleanup(self):
         """Clean up processor resources."""
diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py
index 97098ffed..9e49cb122 100644
--- a/src/pipecat/processors/metrics/frame_processor_metrics.py
+++ b/src/pipecat/processors/metrics/frame_processor_metrics.py
@@ -20,7 +20,6 @@ from pipecat.metrics.metrics import (
     TTFBMetricsData,
     TTSUsageMetricsData,
 )
-from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -40,36 +39,12 @@ class FrameProcessorMetrics(BaseObject):
         processing times, and usage statistics.
         """
         super().__init__()
-        self._task_manager = None
         self._start_ttfb_time = 0
         self._start_processing_time = 0
         self._start_text_aggregation_time = 0
         self._last_ttfb_time = 0
         self._should_report_ttfb = True
 
-    async def setup(self, task_manager: BaseTaskManager):
-        """Set up the metrics collector with a task manager.
-
-        Args:
-            task_manager: The task manager for handling async operations.
-        """
-        self._task_manager = task_manager
-
-    async def cleanup(self):
-        """Clean up metrics collection resources."""
-        await super().cleanup()
-
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Get the associated task manager.
-
-        Returns:
-            The task manager instance for async operations.
-        """
-        if self._task_manager is None:
-            raise RuntimeError("task_manager not set; call setup() first")
-        return self._task_manager
-
     @property
     def ttfb(self) -> float | None:
         """Get the current TTFB value in seconds.
diff --git a/src/pipecat/turns/user_idle_controller.py b/src/pipecat/turns/user_idle_controller.py
index 0fa0053bb..24f358bb8 100644
--- a/src/pipecat/turns/user_idle_controller.py
+++ b/src/pipecat/turns/user_idle_controller.py
@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
-from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -63,29 +62,12 @@ class UserIdleController(BaseObject):
 
         self._user_idle_timeout = user_idle_timeout
 
-        self._task_manager: BaseTaskManager | None = None
-
         self._user_turn_in_progress: bool = False
         self._function_calls_in_progress: int = 0
         self._idle_timer_task: asyncio.Task | None = None
 
         self._register_event_handler("on_user_turn_idle", sync=True)
 
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Returns the configured task manager."""
-        if not self._task_manager:
-            raise RuntimeError(f"{self} user idle controller was not properly setup")
-        return self._task_manager
-
-    async def setup(self, task_manager: BaseTaskManager):
-        """Initialize the controller with the given task manager.
-
-        Args:
-            task_manager: The task manager to be associated with this instance.
-        """
-        self._task_manager = task_manager
-
     async def cleanup(self):
         """Cleanup the controller."""
         await super().cleanup()
@@ -138,17 +120,14 @@ class UserIdleController(BaseObject):
         if self._user_idle_timeout <= 0:
             return
         await self._cancel_idle_timer()
-        self._idle_timer_task = self.task_manager.create_task(
-            self._idle_timer_expired(),
-            f"{self}::idle_timer",
-        )
+        self._idle_timer_task = self.create_task(self._idle_timer_expired())
         # Make sure the task is scheduled.
         await asyncio.sleep(0)
 
     async def _cancel_idle_timer(self):
         """Cancel the idle timer if running."""
         if self._idle_timer_task:
-            await self.task_manager.cancel_task(self._idle_timer_task)
+            await self.cancel_task(self._idle_timer_task)
             self._idle_timer_task = None
 
     async def _idle_timer_expired(self):
diff --git a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py
index b424d7a9e..43100382a 100644
--- a/src/pipecat/turns/user_start/base_user_turn_start_strategy.py
+++ b/src/pipecat/turns/user_start/base_user_turn_start_strategy.py
@@ -11,7 +11,6 @@ from dataclasses import dataclass
 from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.turns.types import ProcessFrameResult
-from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -72,27 +71,11 @@ class BaseUserTurnStartStrategy(BaseObject):
         super().__init__(**kwargs)
         self._enable_interruptions = enable_interruptions
         self._enable_user_speaking_frames = enable_user_speaking_frames
-        self._task_manager: BaseTaskManager | None = None
         self._register_event_handler("on_push_frame", sync=True)
         self._register_event_handler("on_broadcast_frame", sync=True)
         self._register_event_handler("on_user_turn_started", sync=True)
         self._register_event_handler("on_reset_aggregation", sync=True)
 
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Returns the configured task manager."""
-        if not self._task_manager:
-            raise RuntimeError(f"{self} user turn start strategy was not properly setup")
-        return self._task_manager
-
-    async def setup(self, task_manager: BaseTaskManager):
-        """Initialize the strategy with the given task manager.
-
-        Args:
-            task_manager: The task manager to be associated with this instance.
-        """
-        self._task_manager = task_manager
-
     async def cleanup(self):
         """Cleanup the strategy."""
         pass
diff --git a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
index 6f303282c..4bd9709fc 100644
--- a/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
+++ b/src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
@@ -11,7 +11,6 @@ from dataclasses import dataclass
 from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.turns.types import ProcessFrameResult
-from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
 
@@ -70,27 +69,11 @@ class BaseUserTurnStopStrategy(BaseObject):
         """
         super().__init__(**kwargs)
         self._enable_user_speaking_frames = enable_user_speaking_frames
-        self._task_manager: BaseTaskManager | None = None
         self._register_event_handler("on_push_frame", sync=True)
         self._register_event_handler("on_broadcast_frame", sync=True)
         self._register_event_handler("on_user_turn_inference_triggered", sync=True)
         self._register_event_handler("on_user_turn_stopped", sync=True)
 
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Returns the configured task manager."""
-        if not self._task_manager:
-            raise RuntimeError(f"{self} user turn stop strategy was not properly setup")
-        return self._task_manager
-
-    async def setup(self, task_manager: BaseTaskManager):
-        """Initialize the strategy with the given task manager.
-
-        Args:
-            task_manager: The task manager to be associated with this instance.
-        """
-        self._task_manager = task_manager
-
     async def cleanup(self):
         """Cleanup the strategy."""
         pass
diff --git a/src/pipecat/turns/user_turn_controller.py b/src/pipecat/turns/user_turn_controller.py
index 974a8ff0f..51823e368 100644
--- a/src/pipecat/turns/user_turn_controller.py
+++ b/src/pipecat/turns/user_turn_controller.py
@@ -92,8 +92,6 @@ class UserTurnController(BaseObject):
         self._user_turn_strategies = user_turn_strategies
         self._user_turn_stop_timeout = user_turn_stop_timeout
 
-        self._task_manager: BaseTaskManager | None = None
-
         self._user_speaking = False
 
         self._user_turn = False
@@ -108,25 +106,17 @@ class UserTurnController(BaseObject):
         self._register_event_handler("on_user_turn_stop_timeout", sync=True)
         self._register_event_handler("on_reset_aggregation", sync=True)
 
-    @property
-    def task_manager(self) -> BaseTaskManager:
-        """Returns the configured task manager."""
-        if not self._task_manager:
-            raise RuntimeError(f"{self} user turn controller was not properly setup")
-        return self._task_manager
-
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the controller with the given task manager.
 
         Args:
             task_manager: The task manager to be associated with this instance.
         """
-        self._task_manager = task_manager
+        await super().setup(task_manager)
 
         if not self._user_turn_stop_timeout_task:
-            self._user_turn_stop_timeout_task = self.task_manager.create_task(
-                self._user_turn_stop_timeout_task_handler(),
-                f"{self}::_user_turn_stop_timeout_task_handler",
+            self._user_turn_stop_timeout_task = self.create_task(
+                self._user_turn_stop_timeout_task_handler()
             )
 
         await self._setup_strategies()
@@ -136,7 +126,7 @@ class UserTurnController(BaseObject):
         await super().cleanup()
 
         if self._user_turn_stop_timeout_task:
-            await self.task_manager.cancel_task(self._user_turn_stop_timeout_task)
+            await self.cancel_task(self._user_turn_stop_timeout_task)
             self._user_turn_stop_timeout_task = None
 
         await self._cleanup_strategies()
diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py
index 896c62b6a..6733d9106 100644
--- a/src/pipecat/utils/base_object.py
+++ b/src/pipecat/utils/base_object.py
@@ -15,11 +15,13 @@ import asyncio
 import inspect
 import traceback
 from abc import ABC
+from collections.abc import Coroutine
 from dataclasses import dataclass
 from typing import Any
 
 from loguru import logger
 
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.utils import obj_count, obj_id
 
 
@@ -69,6 +71,10 @@ class BaseObject(ABC):
         # event tasks still being executed.
         self._event_tasks = set()
 
+        # Task manager. Populated by setup(); accessing the task_manager
+        # property before setup raises.
+        self._task_manager: BaseTaskManager | None = None
+
     @property
     def id(self) -> int:
         """Get the unique identifier for this object.
@@ -87,6 +93,65 @@ class BaseObject(ABC):
         """
         return self._name
 
+    @property
+    def task_manager(self) -> BaseTaskManager:
+        """Get the task manager for this object.
+
+        Returns:
+            The task manager instance.
+
+        Raises:
+            Exception: If the task manager is not initialized.
+        """
+        if not self._task_manager:
+            raise Exception(f"{self}: TaskManager is not initialized.")
+        return self._task_manager
+
+    async def setup(self, task_manager: BaseTaskManager):
+        """Wire the object up with a task manager.
+
+        Owners of a :class:`BaseObject` should call this on their child objects
+        to propagate the task manager down. Subclasses that own other
+        :class:`BaseObject` instances should override and forward::
+
+            async def setup(self, task_manager):
+                await super().setup(task_manager)
+                await self._child.setup(task_manager)
+
+        Args:
+            task_manager: The task manager to associate with this instance.
+        """
+        self._task_manager = task_manager
+
+    def create_task(self, coroutine: Coroutine, name: str | None = None) -> asyncio.Task:
+        """Create a new task managed by this object's task manager.
+
+        Args:
+            coroutine: The coroutine to run in the task.
+            name: Optional name for the task.
+
+        Returns:
+            The created asyncio task.
+        """
+        if name:
+            name = f"{self}::{name}"
+        else:
+            name = f"{self}::{coroutine.cr_code.co_name}"
+        return self.task_manager.create_task(coroutine, name)
+
+    async def cancel_task(self, task: asyncio.Task, timeout: float | None = 1.0):
+        """Cancel a task managed by this object's task manager.
+
+        A default timeout of 1 second is used in order to avoid potential
+        freezes caused by certain libraries that swallow
+        :class:`asyncio.CancelledError`.
+
+        Args:
+            task: The task to cancel.
+            timeout: Optional timeout for task cancellation.
+        """
+        await self.task_manager.cancel_task(task, timeout)
+
     async def cleanup(self):
         """Clean up resources and wait for running event handlers to complete.
 

From dc035df0aa69e8e8afe57e23216fe0abddf13371 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 13:44:59 -0700
Subject: [PATCH 2/7] Use inherited create_task/cancel_task in PipelineTask

PipelineTask owns its TaskManager but is itself a BaseObject, so it
inherits create_task/cancel_task. Replace the explicit
self._task_manager.create_task(coro, f"{self}::name") call sites with
self.create_task(coro, "name") for consistency with other BaseObject
subclasses.
---
 src/pipecat/pipeline/task.py | 24 ++++++++++++------------
 1 file changed, 12 insertions(+), 12 deletions(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 2783da6ba..479aaada9 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -654,32 +654,32 @@ class PipelineTask(BasePipelineTask):
 
     async def _create_tasks(self):
         """Create and start all pipeline processing tasks."""
-        self._process_push_task = self._task_manager.create_task(
-            self._process_push_queue(), f"{self}::_process_push_queue"
+        self._process_push_task = self.create_task(
+            self._process_push_queue(), "_process_push_queue"
         )
         return self._process_push_task
 
     def _maybe_start_heartbeat_tasks(self):
         """Start heartbeat tasks if heartbeats are enabled and not already running."""
         if self._params.enable_heartbeats and self._heartbeat_push_task is None:
-            self._heartbeat_push_task = self._task_manager.create_task(
-                self._heartbeat_push_handler(), f"{self}::_heartbeat_push_handler"
+            self._heartbeat_push_task = self.create_task(
+                self._heartbeat_push_handler(), "_heartbeat_push_handler"
             )
-            self._heartbeat_monitor_task = self._task_manager.create_task(
-                self._heartbeat_monitor_handler(), f"{self}::_heartbeat_monitor_handler"
+            self._heartbeat_monitor_task = self.create_task(
+                self._heartbeat_monitor_handler(), "_heartbeat_monitor_handler"
             )
 
     def _maybe_start_idle_task(self):
         """Start idle monitoring task if idle timeout is configured."""
         if self._idle_timeout_secs:
-            self._idle_monitor_task = self._task_manager.create_task(
-                self._idle_monitor_handler(), f"{self}::_idle_monitor_handler"
+            self._idle_monitor_task = self.create_task(
+                self._idle_monitor_handler(), "_idle_monitor_handler"
             )
 
     async def _cancel_tasks(self):
         """Cancel all running pipeline tasks."""
         if self._process_push_task:
-            await self._task_manager.cancel_task(self._process_push_task)
+            await self.cancel_task(self._process_push_task)
             self._process_push_task = None
 
         await self._maybe_cancel_heartbeat_tasks()
@@ -691,17 +691,17 @@ class PipelineTask(BasePipelineTask):
             return
 
         if self._heartbeat_push_task:
-            await self._task_manager.cancel_task(self._heartbeat_push_task)
+            await self.cancel_task(self._heartbeat_push_task)
             self._heartbeat_push_task = None
 
         if self._heartbeat_monitor_task:
-            await self._task_manager.cancel_task(self._heartbeat_monitor_task)
+            await self.cancel_task(self._heartbeat_monitor_task)
             self._heartbeat_monitor_task = None
 
     async def _maybe_cancel_idle_task(self):
         """Cancel idle monitoring task if it is running."""
         if self._idle_monitor_task:
-            await self._task_manager.cancel_task(self._idle_monitor_task)
+            await self.cancel_task(self._idle_monitor_task)
             self._idle_monitor_task = None
 
     def _initial_metrics_frame(self) -> MetricsFrame:

From 33db71ec32db8c892197652f7c167075c83540c4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 13:51:46 -0700
Subject: [PATCH 3/7] Call super().setup() in PipelineTask to honor BaseObject
 contract

PipelineTask owns its TaskManager (still constructed in __init__ since
TaskObserver needs it eagerly). Adding the explicit
`await super().setup(self._task_manager)` in `_setup()` formalizes the
BaseObject lifecycle so any future wiring added to BaseObject.setup is
picked up automatically.
---
 src/pipecat/pipeline/task.py | 22 ++++++++--------------
 1 file changed, 8 insertions(+), 14 deletions(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 479aaada9..38b8dd03d 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -654,27 +654,19 @@ class PipelineTask(BasePipelineTask):
 
     async def _create_tasks(self):
         """Create and start all pipeline processing tasks."""
-        self._process_push_task = self.create_task(
-            self._process_push_queue(), "_process_push_queue"
-        )
+        self._process_push_task = self.create_task(self._process_push_queue())
         return self._process_push_task
 
     def _maybe_start_heartbeat_tasks(self):
         """Start heartbeat tasks if heartbeats are enabled and not already running."""
         if self._params.enable_heartbeats and self._heartbeat_push_task is None:
-            self._heartbeat_push_task = self.create_task(
-                self._heartbeat_push_handler(), "_heartbeat_push_handler"
-            )
-            self._heartbeat_monitor_task = self.create_task(
-                self._heartbeat_monitor_handler(), "_heartbeat_monitor_handler"
-            )
+            self._heartbeat_push_task = self.create_task(self._heartbeat_push_handler())
+            self._heartbeat_monitor_task = self.create_task(self._heartbeat_monitor_handler())
 
     def _maybe_start_idle_task(self):
         """Start idle monitoring task if idle timeout is configured."""
         if self._idle_timeout_secs:
-            self._idle_monitor_task = self.create_task(
-                self._idle_monitor_handler(), "_idle_monitor_handler"
-            )
+            self._idle_monitor_task = self.create_task(self._idle_monitor_handler())
 
     async def _cancel_tasks(self):
         """Cancel all running pipeline tasks."""
@@ -759,12 +751,14 @@ class PipelineTask(BasePipelineTask):
 
     async def _setup(self, params: PipelineTaskParams):
         """Set up the pipeline task and all processors."""
+        await super().setup(self._task_manager)
+
         mgr_params = TaskManagerParams(loop=params.loop)
-        self._task_manager.setup(mgr_params)
+        self.task_manager.setup(mgr_params)
 
         setup = FrameProcessorSetup(
             clock=self._clock,
-            task_manager=self._task_manager,
+            task_manager=self.task_manager,
             observer=self._observer,
             pipeline_task=self,
             # Populate the deprecated `tool_resources` field for backwards

From 784667bad22af7c62fc5da9219a589abe1c28f7e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 13:44:59 -0700
Subject: [PATCH 4/7] Use inherited create_task/cancel_task in PipelineTask

PipelineTask owns its TaskManager but is itself a BaseObject, so it
inherits create_task/cancel_task. Replace the explicit
self._task_manager.create_task(coro, f"{self}::name") call sites with
self.create_task(coro, "name") for consistency with other BaseObject
subclasses.
---
 src/pipecat/pipeline/task.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 38b8dd03d..8e35a240a 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -1013,7 +1013,7 @@ class PipelineTask(BasePipelineTask):
 
     def _print_dangling_tasks(self):
         """Log any dangling tasks that haven't been properly cleaned up."""
-        tasks = [t.get_name() for t in self._task_manager.current_tasks()]
+        tasks = [t.get_name() for t in self.task_manager.current_tasks()]
         if tasks:
             logger.warning(f"{self} dangling tasks detected: {tasks}")
 

From 15531c8112cdec86ecf7f8d1f21ca46c6c92c66c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 14:00:46 -0700
Subject: [PATCH 5/7] Wire TaskObserver via setup() instead of constructor

TaskObserver previously took a TaskManager in __init__ and reached into
it directly. Since BaseObject now provides task_manager / create_task /
cancel_task, drop the constructor argument and call
`observer.setup(task_manager)` from PipelineTask._setup() before
starting it.
---
 src/pipecat/pipeline/task.py          |  7 ++++---
 src/pipecat/pipeline/task_observer.py | 13 ++++---------
 2 files changed, 8 insertions(+), 12 deletions(-)

diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py
index 8e35a240a..51b431e86 100644
--- a/src/pipecat/pipeline/task.py
+++ b/src/pipecat/pipeline/task.py
@@ -303,7 +303,7 @@ class PipelineTask(BasePipelineTask):
 
         # This task maneger will handle all the asyncio tasks created by this
         # PipelineTask and its frame processors.
-        self._task_manager = task_manager or TaskManager()
+        self._pipeline_task_manager = task_manager or TaskManager()
 
         # This queue is the queue used to push frames to the pipeline.
         self._push_queue = asyncio.Queue()
@@ -386,7 +386,7 @@ class PipelineTask(BasePipelineTask):
         # The task observer acts as a proxy to the provided observers. This way,
         # we only need to pass a single observer (using the StartFrame) which
         # then just acts as a proxy.
-        self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)
+        self._observer = TaskObserver(observers=observers)
 
         # These events can be used to check which frames make it to the source
         # or sink processors. Instead of calling the event handlers for every
@@ -751,7 +751,7 @@ class PipelineTask(BasePipelineTask):
 
     async def _setup(self, params: PipelineTaskParams):
         """Set up the pipeline task and all processors."""
-        await super().setup(self._task_manager)
+        await super().setup(self._pipeline_task_manager)
 
         mgr_params = TaskManagerParams(loop=params.loop)
         self.task_manager.setup(mgr_params)
@@ -774,6 +774,7 @@ class PipelineTask(BasePipelineTask):
         await self._load_setup_files()
 
         # Start task observer.
+        await self._observer.setup(self.task_manager)
         await self._observer.start()
 
     async def _cleanup(self, cleanup_pipeline: bool):
diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py
index 5b29850a6..819de93ed 100644
--- a/src/pipecat/pipeline/task_observer.py
+++ b/src/pipecat/pipeline/task_observer.py
@@ -17,7 +17,6 @@ from typing import Any
 from attr import dataclass
 
 from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
-from pipecat.utils.asyncio.task_manager import BaseTaskManager
 
 
 @dataclass
@@ -62,19 +61,16 @@ class TaskObserver(BaseObserver):
         self,
         *,
         observers: list[BaseObserver] | None = None,
-        task_manager: BaseTaskManager,
         **kwargs,
     ):
         """Initialize the TaskObserver.
 
         Args:
             observers: List of observers to manage. Defaults to empty list.
-            task_manager: Task manager for creating and managing observer tasks.
             **kwargs: Additional arguments passed to the base observer.
         """
         super().__init__(**kwargs)
         self._observers = observers or []
-        self._task_manager = task_manager
         self._proxies: dict[BaseObserver, Proxy] | None = (
             None  # Becomes a dict after start() is called
         )
@@ -106,7 +102,7 @@ class TaskObserver(BaseObserver):
             # Remove the proxy so it doesn't get called anymore.
             del self._proxies[observer]
             # Cancel the proxy task right away.
-            await self._task_manager.cancel_task(proxy.task)
+            await self.cancel_task(proxy.task)
 
         # Remove the observer from the list.
         if observer in self._observers:
@@ -122,7 +118,7 @@ class TaskObserver(BaseObserver):
             return
 
         for proxy in self._proxies.values():
-            await self._task_manager.cancel_task(proxy.task)
+            await self.cancel_task(proxy.task)
 
     async def cleanup(self):
         """Cleanup all proxy observers."""
@@ -157,9 +153,8 @@ class TaskObserver(BaseObserver):
     def _create_proxy(self, observer: BaseObserver) -> Proxy:
         """Create a proxy for a single observer."""
         queue = asyncio.Queue()
-        task = self._task_manager.create_task(
-            self._proxy_task_handler(queue, observer),
-            f"TaskObserver::{observer}::_proxy_task_handler",
+        task = self.create_task(
+            self._proxy_task_handler(queue, observer), f"{observer}::_proxy_task_handler"
         )
         proxy = Proxy(queue=queue, task=task, observer=observer)
         return proxy

From 4f85e7c089aaa8c5bbfacb963ce8f0faa3ffa56b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 14:36:16 -0700
Subject: [PATCH 6/7] Fix pyright cr_code access on Coroutine in
 BaseObject.create_task

`collections.abc.Coroutine` doesn't expose `cr_code`/`co_name`; only
native coroutine objects do. Use `getattr` chains so pyright is happy
and any non-native awaitable falls back to a generic task name instead
of crashing.
---
 src/pipecat/utils/base_object.py | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py
index 6733d9106..af68abb59 100644
--- a/src/pipecat/utils/base_object.py
+++ b/src/pipecat/utils/base_object.py
@@ -133,11 +133,12 @@ class BaseObject(ABC):
         Returns:
             The created asyncio task.
         """
-        if name:
-            name = f"{self}::{name}"
-        else:
-            name = f"{self}::{coroutine.cr_code.co_name}"
-        return self.task_manager.create_task(coroutine, name)
+        if not name:
+            # Native coroutines expose ``cr_code``; fall back to a generic
+            # name for any other awaitable subtype.
+            cr_code = getattr(coroutine, "cr_code", None)
+            name = getattr(cr_code, "co_name", "task")
+        return self.task_manager.create_task(coroutine, f"{self}::{name}")
 
     async def cancel_task(self, task: asyncio.Task, timeout: float | None = 1.0):
         """Cancel a task managed by this object's task manager.

From 77058b01c42ef9eca53162a87a940fc33655c75d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= <aleix@daily.co>
Date: Fri, 8 May 2026 13:54:06 -0700
Subject: [PATCH 7/7] Add changelog for #4449

---
 changelog/4449.changed.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 changelog/4449.changed.md

diff --git a/changelog/4449.changed.md b/changelog/4449.changed.md
new file mode 100644
index 000000000..aef304037
--- /dev/null
+++ b/changelog/4449.changed.md
@@ -0,0 +1 @@
+- Moved `create_task`, `cancel_task`, the `task_manager` property, and `setup(task_manager)` up from `FrameProcessor` to `BaseObject`. Custom `BaseObject` subclasses (turn strategies, controllers, etc.) now inherit these methods directly instead of reimplementing the task manager wiring. Owners propagate the task manager to their child `BaseObject`s via `await child.setup(task_manager)`.