Separate bus messages from pipeline frames

BusMessage was a mixin tacked onto DataFrame / SystemFrame so the bus
could reuse the frame priority machinery. That made every bus message
also a Frame, which is misleading — bus messages travel on the bus, not
through pipelines. If a worker actually needs to ship a frame, it wraps
it in BusFrameMessage.

BusMessage is now a plain dataclass base carrying source/target.
BusDataMessage and BusSystemMessage are empty subclasses that exist
only as priority markers. The bus router and the priority queue check
``isinstance(item, BusSystemMessage)`` directly instead of
``isinstance(item, SystemFrame)``.

The serializer test that round-tripped DataFrame.name (a non-init
field) is rewritten against a local _MessageWithNonInit(BusDataMessage)
subclass so the serializer's init=False path stays covered.

src/pipecat/bus/bus.py

@@ -15,10 +15,9 @@ from abc import abstractmethod
 from dataclasses import dataclass, field
 from typing import cast
 
-from pipecat.bus.messages import BusLocalMessage, BusMessage
+from pipecat.bus.messages import BusLocalMessage, BusMessage, BusSystemMessage
 from pipecat.bus.queue import BusMessageQueue
 from pipecat.bus.subscriber import BusSubscriber
-from pipecat.frames.frames import SystemFrame
 from pipecat.utils.base_object import BaseObject
 
 
@@ -167,7 +166,7 @@ class WorkerBus(BaseObject):
         try:
             while True:
                 message = await sub.queue.get()
-                if isinstance(message, SystemFrame):
+                if isinstance(message, BusSystemMessage):
                     await sub.subscriber.on_bus_message(cast(BusMessage, message))
                 else:
                     sub.data_queue.put_nowait(message)

src/pipecat/bus/messages.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from pipecat.frames.frames import DataFrame, Frame, SystemFrame
+from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.registry.types import WorkerRegistryEntry
 
@@ -28,11 +28,19 @@ if TYPE_CHECKING:
 # ---------------------------------------------------------------------------
 
 
+@dataclass(kw_only=True)
 class BusMessage:
-    """Mixin carrying source/target metadata for bus messages.
+    """Base class for messages carried by the `WorkerBus`.
 
-    Not a frame itself. Combined with `DataFrame` or `SystemFrame`
+    Bus messages are independent of pipeline `Frame`s — if a worker needs
-    to create concrete message types with appropriate priority.
+    to ship a frame between pipelines it wraps it in a `BusFrameMessage`.
+    Subclasses choose delivery priority by extending :class:`BusDataMessage`
+    (normal priority, FIFO) or :class:`BusSystemMessage` (high priority,
+    delivered ahead of queued data messages).
+
+    Parameters:
+        source: Name of the worker or component that sent this message.
+        target: Name of the intended recipient worker, or None for broadcast.
     """
 
     source: str
@@ -49,29 +57,24 @@ class BusLocalMessage:
 
 
 @dataclass(kw_only=True)
-class BusDataMessage(BusMessage, DataFrame):
+class BusDataMessage(BusMessage):
     """Normal-priority bus message.
 
-    Parameters:
+    Delivered in FIFO order on the subscriber's data queue.
-        source: Name of the worker or component that sent this message.
-        target: Name of the intended recipient worker, or None for broadcast.
     """
 
-    source: str
+    pass
-    target: str | None = None
 
 
 @dataclass(kw_only=True)
-class BusSystemMessage(BusMessage, SystemFrame):
+class BusSystemMessage(BusMessage):
-    """High-priority bus message that preempts normal messages in subscriber queues.
+    """High-priority bus message.
 
-    Parameters:
+    Delivered ahead of any queued :class:`BusDataMessage` on the
-        source: Name of the worker or component that sent this message.
+    subscriber's priority queue.
-        target: Name of the intended recipient worker, or None for broadcast.
     """
 
-    source: str
+    pass
-    target: str | None = None
 
 
 # ---------------------------------------------------------------------------

src/pipecat/bus/queue.py

@@ -9,7 +9,7 @@
 import asyncio
 from typing import Any
 
-from pipecat.frames.frames import SystemFrame
+from pipecat.bus.messages import BusSystemMessage
 
 HIGH_PRIORITY = 1
 LOW_PRIORITY = 2
@@ -18,9 +18,9 @@ LOW_PRIORITY = 2
 class BusMessageQueue(asyncio.PriorityQueue):
     """Priority queue that delivers system messages before normal messages.
 
-    Messages that extend ``SystemFrame`` (e.g. cancel messages) get high
+    Messages that extend :class:`BusSystemMessage` (e.g. cancel messages)
-    priority. All other messages are delivered in FIFO order at normal
+    get high priority. All other messages are delivered in FIFO order at
-    priority.
+    normal priority.
     """
 
     def __init__(self):
@@ -35,7 +35,7 @@ class BusMessageQueue(asyncio.PriorityQueue):
         Args:
             item: The bus message to enqueue.
         """
-        if isinstance(item, SystemFrame):
+        if isinstance(item, BusSystemMessage):
             self._high_counter += 1
             super().put_nowait((HIGH_PRIORITY, self._high_counter, item))
         else:
@@ -48,7 +48,7 @@ class BusMessageQueue(asyncio.PriorityQueue):
         Args:
             item: The bus message to enqueue.
         """
-        if isinstance(item, SystemFrame):
+        if isinstance(item, BusSystemMessage):
             self._high_counter += 1
             await super().put((HIGH_PRIORITY, self._high_counter, item))
         else:

tests/test_serializers.py

@@ -5,6 +5,7 @@
 #
 
 import unittest
+from dataclasses import dataclass, field
 
 from pydantic import BaseModel
 
@@ -35,6 +36,11 @@ class _UserInfo(BaseModel):
     address: _Address | None = None
 
 
+@dataclass(kw_only=True)
+class _MessageWithNonInit(BusDataMessage):
+    tag: str = field(init=False, default="default")
+
+
 class TestJSONMessageSerializer(unittest.TestCase):
     def setUp(self):
         self.serializer = JSONMessageSerializer()
@@ -232,14 +238,13 @@ class TestJSONMessageSerializer(unittest.TestCase):
 
     def test_non_init_fields_preserved(self):
         """Non-init dataclass fields survive round-trip via setattr."""
-        msg = BusDataMessage(source="task_a", target="task_b")
+        msg = _MessageWithNonInit(source="worker_a", target="worker_b")
-        # name is a non-init field on DataFrame
+        msg.tag = "custom_tag"
-        msg.name = "custom_name"
 
         data = self.serializer.serialize(msg)
         restored = self.serializer.deserialize(data)
 
-        self.assertEqual(restored.name, "custom_name")
+        self.assertEqual(restored.tag, "custom_tag")
 
 
 if __name__ == "__main__":