Replace set_registry with attach(*, registry, bus) on BaseTask

`BaseTask` no longer takes `bus=` in its constructor. Instead
the runner now hands both the registry and the bus to a task via
`task.attach(registry=..., bus=...)` (called from
`PipelineRunner.spawn()`), and `bus` / `registry` are
properties that raise if accessed before attach. `PipelineTask`,
`LLMTask`, and `LLMContextTask` lose their `bus=` parameters
to match, and `_BusEdgeProcessor` now stores only a task
reference and reads `task.bus` lazily so bridged pipelines work
even though the bus isn't known at construction time.

src/pipecat/bus/bridge_processor.py

@@ -161,7 +161,6 @@ class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
     def __init__(
         self,
         *,
-        bus: TaskBus,
         task: "BaseTask",
         direction: FrameDirection,
         bridges: tuple[str, ...] = (),
@@ -171,9 +170,11 @@ class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
         """Initialize the edge processor.
 
         Args:
-            bus: The ``TaskBus`` to exchange frames with.
+            task: The owning task; the edge reads ``task.bus`` lazily
-            task: The owning task; ``task.name`` is the message source
+                so the bus only needs to be set (via
-                and ``task.active`` gates inbound frames.
+                :meth:`BaseTask.attach`) by the time the processor is
+                set up. ``task.name`` is the message source and
+                ``task.active`` gates inbound frames.
             direction: Direction this edge captures and forwards to the
                 bus. Inbound frames from the bus travelling in the
                 opposite direction are injected here.
@@ -184,7 +185,6 @@ class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
             **kwargs: Additional arguments passed to ``FrameProcessor``.
         """
         super().__init__(**kwargs)
-        self._bus = bus
         self._task = task
         self._direction = direction
         self._bridges = bridges
@@ -193,12 +193,12 @@ class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
     async def setup(self, setup: FrameProcessorSetup):
         """Subscribe to the bus during processor setup."""
         await super().setup(setup)
-        await self._bus.subscribe(self)
+        await self._task.bus.subscribe(self)
 
     async def cleanup(self):
         """Unsubscribe from the bus on cleanup."""
         await super().cleanup()
-        await self._bus.unsubscribe(self)
+        await self._task.bus.unsubscribe(self)
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Pass the frame through locally and forward matching ones to the bus."""
@@ -212,7 +212,7 @@ class _BusEdgeProcessor(FrameProcessor, BusSubscriber):
         if self._exclude_frames and isinstance(frame, self._exclude_frames):
             return
 
-        await self._bus.send(
+        await self._task.bus.send(
             BusFrameMessage(source=self._task.name, frame=frame, direction=direction)
         )
 

src/pipecat/pipeline/base_task.py

@@ -152,7 +152,6 @@ class BaseTask(BaseObject, BusSubscriber):
         self,
         name: str | None = None,
         *,
-        bus: TaskBus | None = None,
         active: bool = True,
     ):
         """Initialize the BaseTask.
@@ -161,15 +160,15 @@ class BaseTask(BaseObject, BusSubscriber):
             name: Unique name for this task. If ``None``, an auto-generated
                 name is used (useful for instances that don't participate
                 in inter-task communication).
-            bus: The `TaskBus` for inter-task communication. ``None``
-                means this task doesn't participate in inter-task
-                messaging; methods that require a bus will fail at
-                call time.
             active: Whether the task starts active. Defaults to True.
         """
         super().__init__(name=name)
 
-        self._bus = bus
+        # Runner-provided context. Populated by ``attach()`` before
+        # ``run()`` is called. Accessing ``self.bus`` / ``self.registry``
+        # before ``attach()`` raises.
+        self._bus: TaskBus | None = None
+        self._registry: TaskRegistry | None = None
 
         # Activation. Pending activation is deferred until the task
         # starts, then on_activated fires.
@@ -184,9 +183,6 @@ class BaseTask(BaseObject, BusSubscriber):
         self._started_at: float | None = None
         self._finished_event: asyncio.Event = asyncio.Event()
 
-        # Registry is set by the runner via set_registry().
-        self._registry: TaskRegistry | None = None
-
         # Job coordination. Worker state tracks active job requests
         # keyed by job_id, supporting multiple jobs in flight
         # (e.g. parallel @job handlers). Each running handler has a
@@ -223,8 +219,14 @@ class BaseTask(BaseObject, BusSubscriber):
         self._register_event_handler("on_task_failed")
 
     @property
-    def bus(self) -> TaskBus | None:
+    def bus(self) -> TaskBus:
-        """The bus instance for task communication."""
+        """The bus this task is attached to.
+
+        Raises:
+            RuntimeError: If accessed before :meth:`attach` has been called.
+        """
+        if self._bus is None:
+            raise RuntimeError(f"Task '{self}': bus is not set; call attach() first.")
         return self._bus
 
     @property
@@ -243,8 +245,14 @@ class BaseTask(BaseObject, BusSubscriber):
         return self._parent
 
     @property
-    def registry(self) -> TaskRegistry | None:
+    def registry(self) -> TaskRegistry:
-        """The shared task registry, if set by a runner."""
+        """The shared task registry this task is attached to.
+
+        Raises:
+            RuntimeError: If accessed before :meth:`attach` has been called.
+        """
+        if self._registry is None:
+            raise RuntimeError(f"Task '{self}': registry is not set; call attach() first.")
         return self._registry
 
     @property
@@ -276,13 +284,19 @@ class BaseTask(BaseObject, BusSubscriber):
         """Active job groups launched by this task, keyed by job_id."""
         return self._job_groups
 
-    def set_registry(self, registry: TaskRegistry) -> None:
+    def attach(self, *, registry: TaskRegistry, bus: TaskBus) -> None:
-        """Set the shared task registry.
+        """Attach the task to a runner-provided registry and bus.
+
+        Called by the runner (typically from ``spawn()``) before the
+        task is run. After this call, :attr:`registry` and :attr:`bus`
+        return the provided instances.
 
         Args:
-            registry: The shared registry instance.
+            registry: The shared task registry.
+            bus: The shared task bus.
         """
         self._registry = registry
+        self._bus = bus
 
     async def cleanup(self) -> None:
         """Clean up the task and release resources.

src/pipecat/pipeline/runner.py

@@ -159,7 +159,7 @@ class PipelineRunner(BaseObject, BusSubscriber):
         if task.name in self._entries:
             logger.error(f"PipelineRunner '{self}': task '{task.name}' already exists, skipping")
             return
-        task.set_registry(self._registry)
+        task.attach(registry=self._registry, bus=self._bus)
         await self._registry.watch(task.name, self._on_local_task_ready)
         entry = _TaskEntry(task=task)
         self._entries[task.name] = entry

src/pipecat/pipeline/task.py

@@ -208,7 +208,6 @@ class PipelineTask(BaseTask):
         *,
         additional_span_attributes: dict | None = None,
         app_resources: Any = None,
-        bus: TaskBus | None = None,
         bridged: tuple[str, ...] | None = None,
         cancel_on_idle_timeout: bool = True,
         cancel_timeout_secs: float = CANCEL_TIMEOUT_SECS,
@@ -243,14 +242,12 @@ class PipelineTask(BaseTask):
                 ``FunctionCallParams.app_resources``. The framework never
                 copies or clears this object; the caller retains their handle
                 and can read any mutations after the task finishes.
-            bus: Optional :class:`~pipecat.bus.TaskBus` for inter-task
-                communication. ``None`` means this task doesn't
-                participate in inter-task messaging.
             bridged: Bridge configuration. ``None`` means the pipeline
                 is not bridged. An empty tuple ``()`` wraps the pipeline
                 with bus edge processors that accept frames from all
                 bridges. A tuple of names like ``("voice",)`` accepts
-                only frames from those bridges. Requires ``bus``.
+                only frames from those bridges. The bus comes from
+                :meth:`attach` (called by the runner).
             cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
                 the idle timeout is reached.
             cancel_timeout_secs: Timeout (in seconds) to wait for cancellation to happen
@@ -281,10 +278,8 @@ class PipelineTask(BaseTask):
                     Use ``app_resources`` instead. ``tool_resources`` will be
                     removed in a future version.
         """
-        super().__init__(name=name, bus=bus)
+        super().__init__(name=name)
         self._bridged = bridged
-        if bridged is not None and bus is None:
-            raise ValueError(f"PipelineTask '{self}': ``bridged`` requires a ``bus`` to be set.")
         if tool_resources is not None:
             with warnings.catch_warnings():
                 warnings.simplefilter("always")
@@ -406,11 +401,11 @@ class PipelineTask(BaseTask):
 
         # 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.
+        # frames are injected back into the local pipeline. The edges
+        # read the task's bus lazily, so the bus only needs to be set
+        # (via ``attach()``) before ``run()`` is called.
         if bridged is not None:
-            assert bus is not None  # validated above
             edge_source = _BusEdgeProcessor(
-                bus=bus,
                 task=self,
                 direction=FrameDirection.UPSTREAM,
                 bridges=bridged,
@@ -418,7 +413,6 @@ class PipelineTask(BaseTask):
                 name=f"{self}::EdgeSource",
             )
             edge_sink = _BusEdgeProcessor(
-                bus=bus,
                 task=self,
                 direction=FrameDirection.DOWNSTREAM,
                 bridges=bridged,

src/pipecat/tasks/llm/llm_context_task.py

@@ -11,7 +11,6 @@ self-contained conversation context, removing the need for subclasses
 to manually wire `LLMContextAggregatorPair`.
 """
 
-from pipecat.bus import TaskBus
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.llm_response_universal import (
@@ -38,7 +37,6 @@ class LLMContextTask(LLMTask):
 
         task = LLMContextTask(
             "worker",
-            bus=bus,
             llm=OpenAILLMService(...),
         )
 
@@ -51,7 +49,6 @@ class LLMContextTask(LLMTask):
         self,
         name: str,
         *,
-        bus: TaskBus,
         llm: LLMService,
         active: bool = False,
         bridged: tuple[str, ...] | None = None,
@@ -64,7 +61,6 @@ class LLMContextTask(LLMTask):
 
         Args:
             name: Unique name for this task.
-            bus: The `TaskBus` for inter-task communication.
             llm: The LLM service.
             active: Whether the task starts active. Defaults to False.
             bridged: Bridge configuration forwarded to ``PipelineTask``.
@@ -95,7 +91,6 @@ class LLMContextTask(LLMTask):
 
         super().__init__(
             name,
-            bus=bus,
             llm=llm,
             pipeline=pipeline,
             active=active,

src/pipecat/tasks/llm/llm_task.py

@@ -18,7 +18,6 @@ from dataclasses import dataclass
 from typing import Any
 
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
-from pipecat.bus import TaskBus
 from pipecat.frames.frames import (
     ControlFrame,
     Frame,
@@ -79,8 +78,8 @@ class LLMTask(PipelineTask):
         self,
         name: str,
         *,
-        bus: TaskBus,
         llm: LLMService,
+        pipeline: Pipeline | None = None,
         active: bool = False,
         bridged: tuple[str, ...] | None = None,
         defer_tool_frames: bool = True,
@@ -89,9 +88,12 @@ class LLMTask(PipelineTask):
 
         Args:
             name: Unique name for this task.
-            bus: The `TaskBus` for inter-task communication.
             llm: The LLM service. ``@tool`` decorated methods are
                 automatically registered on it.
+            pipeline: Optional pipeline override. When ``None``,
+                defaults to ``Pipeline([llm])``. Subclasses can pass a
+                custom pipeline that wraps the LLM with additional
+                processors.
             active: Whether the task starts active. Defaults to False.
             bridged: Bridge configuration forwarded to ``PipelineTask``.
                 Pass ``()`` to wrap the LLM pipeline with bus edge
@@ -111,12 +113,11 @@ class LLMTask(PipelineTask):
         self._llm = llm
         self._register_tools(llm)
 
-        pipeline = Pipeline([self._llm])
+        pipeline = pipeline if pipeline is not None else Pipeline([self._llm])
 
         super().__init__(
             pipeline,
             name=name,
-            bus=bus,
             bridged=bridged,
             exclude_frames=(PipelineFlushFrame,),
             enable_rtvi=False,