From 7506af586129e30db52bf8e84d742b7289cecd0b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 13 May 2026 21:04:53 -0700 Subject: [PATCH] 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 | 16 ++++---- src/pipecat/pipeline/base_task.py | 46 +++++++++++++++-------- src/pipecat/pipeline/runner.py | 2 +- src/pipecat/pipeline/task.py | 18 +++------ src/pipecat/tasks/llm/llm_context_task.py | 5 --- src/pipecat/tasks/llm/llm_task.py | 11 +++--- 6 files changed, 51 insertions(+), 47 deletions(-) diff --git a/src/pipecat/bus/bridge_processor.py b/src/pipecat/bus/bridge_processor.py index da8690c56..bb6af249d 100644 --- a/src/pipecat/bus/bridge_processor.py +++ b/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; ``task.name`` is the message source - and ``task.active`` gates inbound frames. + task: The owning task; the edge reads ``task.bus`` lazily + so the bus only needs to be set (via + :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) ) diff --git a/src/pipecat/pipeline/base_task.py b/src/pipecat/pipeline/base_task.py index 55207043e..1440124c8 100644 --- a/src/pipecat/pipeline/base_task.py +++ b/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: - """The bus instance for task communication.""" + def bus(self) -> TaskBus: + """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: - """The shared task registry, if set by a runner.""" + def registry(self) -> TaskRegistry: + """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: - """Set the shared task registry. + def attach(self, *, registry: TaskRegistry, bus: TaskBus) -> None: + """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. diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 3d40fd8d0..d60493a0b 100644 --- a/src/pipecat/pipeline/runner.py +++ b/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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f6c62df5b..e43f10840 100644 --- a/src/pipecat/pipeline/task.py +++ b/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, diff --git a/src/pipecat/tasks/llm/llm_context_task.py b/src/pipecat/tasks/llm/llm_context_task.py index 265c7b537..6a570e4fb 100644 --- a/src/pipecat/tasks/llm/llm_context_task.py +++ b/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, diff --git a/src/pipecat/tasks/llm/llm_task.py b/src/pipecat/tasks/llm/llm_task.py index 8c770be2d..7b05fd5e5 100644 --- a/src/pipecat/tasks/llm/llm_task.py +++ b/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,