wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add native RTVI⇄bus UI bridge to PipelineWorker

Browse Source 
When RTVI is enabled, PipelineWorker now republishes inbound ui-event /
ui-snapshot / ui-cancel-task messages onto the bus as a broadcast
BusUIEventMessage, and translates outbound BusUICommandMessage / BusUITask*
carriers into the matching RTVI frames. This lets a UIWorker on the bus observe
and drive the client UI with no decorator or manual wiring; when no UIWorker is
present the events are simply unconsumed.

The BusUI* carriers live in the bus layer so both pipeline and workers can
reference them without an import cycle.
This commit is contained in:
Mark Backman
2026-05-21 16:51:22 -04:00
parent ee3d1128ec
commit 02667a7255
3 changed files with 538 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
src/pipecat/bus/ui_messages.py Normal file
View File

@@ -0,0 +1,178 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus carriers for the UI Worker protocol.
These dataclasses are the on-the-bus shape that ``UIWorker`` (see
``pipecat.workers.ui``) and the native RTVI⇄bus bridge in
``PipelineWorker`` exchange. They are NOT the on-the-wire format the
client sees; that lives in ``pipecat.processors.frameworks.rtvi.models``
(``UIEventMessage``, ``UICommandMessage``, ``UITaskMessage``, ...). The
bridge translates between the two.
- ``BusUIEventMessage`` and ``BusUICommandMessage`` carry client
events and server commands respectively.
- ``BusUITaskGroupStartedMessage``, ``BusUITaskUpdateMessage``,
``BusUITaskCompletedMessage``, and ``BusUITaskGroupCompletedMessage``
carry the four phases of a user-facing task group's lifecycle. The
"task" naming here mirrors the fixed RTVI ``ui-task`` wire protocol;
the server-side mechanism that drives them is a job group (see
``UIWorker.user_job_group``).
The carriers live in the ``bus`` layer (rather than alongside
``UIWorker``) because both ``PipelineWorker`` (in ``pipecat.pipeline``)
and ``UIWorker`` (in ``pipecat.workers``) reference them, and
``pipeline`` must not import from ``workers``.
"""
from dataclasses import dataclass
from typing import Any
from pipecat.bus.messages import BusDataMessage
#: Internal ``event_name`` used by the UI bridge when republishing a
#: ``ui-snapshot`` wire message onto the bus as a
#: ``BusUIEventMessage``. ``UIWorker``'s bus dispatch matches on this
#: name to route the snapshot into ``_latest_snapshot`` storage. The
#: leading double underscore marks the name as internal so app-defined
#: ``@on_ui_event`` handlers can't collide with it.
_UI_SNAPSHOT_BUS_EVENT_NAME = "__ui_snapshot"
#: Internal ``event_name`` used by the UI bridge when republishing a
#: ``ui-cancel-task`` wire message onto the bus as a
#: ``BusUIEventMessage``. ``UIWorker``'s bus dispatch matches on this
#: name to route to ``cancel_job_group``. Internal; not part of the
#: public wire format.
_UI_CANCEL_TASK_BUS_EVENT_NAME = "__cancel_task"
@dataclass
class BusUIEventMessage(BusDataMessage):
"""A UI event sent from the client to a server-side worker.
Emitted by the native UI bridge in ``PipelineWorker`` when the
client dispatches an event via
``PipecatClient.sendUIEvent(event, payload)``. ``UIWorker``
subclasses dispatch these to ``@on_ui_event(name)`` handlers.
Parameters:
event_name: App-defined event name.
payload: App-defined payload. Schemaless by design.
"""
event_name: str = ""
payload: Any = None
@dataclass
class BusUICommandMessage(BusDataMessage):
"""A UI command sent from a server-side worker to the client.
Published by ``UIWorker.send_command(name, payload)``. The native
UI bridge in ``PipelineWorker`` translates this to an
``RTVIUICommandFrame(command=command_name, payload=payload)`` and
pushes it through the pipeline.
Parameters:
command_name: App-defined command name.
payload: App-defined payload (already a plain dict by the time
it lands on the bus).
"""
command_name: str = ""
payload: Any = None
# ---------------------------------------------------------------------------
# UI task lifecycle
# ---------------------------------------------------------------------------
@dataclass
class BusUITaskGroupStartedMessage(BusDataMessage):
"""A user-facing task group has been dispatched.
Published by ``UIWorker.user_job_group(...)`` on entry. The bridge
forwards it to the client as a ``ui-task`` envelope with
``kind = "group_started"``.
Parameters:
task_id: Shared task identifier for the group.
agents: Names of the workers the work was dispatched to.
label: Optional human-readable label for the group.
cancellable: Whether the client may request cancellation.
at: Epoch milliseconds when the group started.
"""
task_id: str = ""
agents: list[str] | None = None
label: str | None = None
cancellable: bool = True
at: int = 0
@dataclass
class BusUITaskUpdateMessage(BusDataMessage):
"""Per-task progress for a user-facing task group.
Forwarded by the ``UIWorker`` whenever a worker emits a
``BusJobUpdateMessage`` whose ``job_id`` matches a registered user
task group. The bridge forwards to the client as a ``ui-task``
envelope with ``kind = "task_update"``.
Parameters:
task_id: The shared task identifier.
agent_name: The worker that produced the update.
data: The worker's update payload, forwarded verbatim.
at: Epoch milliseconds when the update was emitted on the bus.
"""
task_id: str = ""
agent_name: str = ""
data: Any = None
at: int = 0
@dataclass
class BusUITaskCompletedMessage(BusDataMessage):
"""A worker in a user-facing task group has completed.
Forwarded by the ``UIWorker`` whenever a worker's
``BusJobResponseMessage`` arrives for a registered user task group.
The bridge forwards to the client as a ``ui-task`` envelope with
``kind = "task_completed"``.
Parameters:
task_id: The shared task identifier.
agent_name: The worker that produced the response.
status: Completion status as a string (``JobStatus`` value).
response: The worker's response payload.
at: Epoch milliseconds when the response was received.
"""
task_id: str = ""
agent_name: str = ""
status: str = ""
response: Any = None
at: int = 0
@dataclass
class BusUITaskGroupCompletedMessage(BusDataMessage):
"""A user-facing task group has completed.
Published when ``UIWorker.user_job_group(...)`` exits, after every
worker has responded (or the group has been cancelled). The bridge
forwards to the client as a ``ui-task`` envelope with
``kind = "group_completed"``.
Parameters:
task_id: The shared task identifier.
at: Epoch milliseconds when the group completed.
"""
task_id: str = ""
at: int = 0

111
src/pipecat/pipeline/worker.py
View File

@@ -22,6 +22,17 @@ from pydantic import BaseModel, ConfigDict, Field
from pipecat.bus import BusCancelWorkerMessage, BusEndWorkerMessage, WorkerBus
from pipecat.bus.bridge_processor import _BusEdgeProcessor
from pipecat.bus.messages import BusMessage
from pipecat.bus.ui_messages import (
_UI_CANCEL_TASK_BUS_EVENT_NAME,
_UI_SNAPSHOT_BUS_EVENT_NAME,
BusUICommandMessage,
BusUIEventMessage,
BusUITaskCompletedMessage,
BusUITaskGroupCompletedMessage,
BusUITaskGroupStartedMessage,
BusUITaskUpdateMessage,
)
from pipecat.clocks.base_clock import BaseClock
from pipecat.clocks.system_clock import SystemClock
from pipecat.frames.frames import (
@@ -52,6 +63,16 @@ from pipecat.pipeline.utils import run_setup_hook
from pipecat.pipeline.worker_observer import WorkerObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor
from pipecat.processors.frameworks.rtvi.frames import RTVIUICommandFrame, RTVIUITaskFrame
from pipecat.processors.frameworks.rtvi.models import (
UICancelTaskMessage,
UIEventMessage,
UISnapshotMessage,
UITaskCompletedData,
UITaskGroupCompletedData,
UITaskGroupStartedData,
UITaskUpdateData,
)
from pipecat.utils.asyncio.task_manager import BaseTaskManager, TaskManager, TaskManagerParams
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.tracing_context import TracingContext
@@ -377,6 +398,34 @@ class PipelineWorker(BaseWorker):
async def on_client_ready(rtvi: RTVIProcessor):
await rtvi.set_bot_ready()
# Republish typed RTVI UI messages from the client onto the bus
# as a single BusUIEventMessage carrier so UIWorker subscribers
# can dispatch them.
@self.rtvi.event_handler("on_ui_message")
async def on_ui_message(rtvi: RTVIProcessor, message):
if isinstance(message, UIEventMessage):
event_name = message.data.event
payload = message.data.payload
elif isinstance(message, UISnapshotMessage):
event_name = _UI_SNAPSHOT_BUS_EVENT_NAME
payload = message.data.tree.model_dump(exclude_none=True)
elif isinstance(message, UICancelTaskMessage):
event_name = _UI_CANCEL_TASK_BUS_EVENT_NAME
payload = {
"task_id": message.data.task_id,
"reason": message.data.reason,
}
else:
return
await self.send_bus_message(
BusUIEventMessage(
source=self.name,
target=None,
event_name=event_name,
payload=payload,
)
)
# This is the idle event. When selected frames are pushed from any
# processor we consider the pipeline is not idle. We use an observer
# which will be listening any part of the pipeline.
@@ -704,6 +753,68 @@ class PipelineWorker(BaseWorker):
for frame in frames:
await self.queue_frame(frame, direction)
async def on_bus_message(self, message: BusMessage) -> None:
"""Handle bus messages for outbound RTVI UI messages.
Runs the base lifecycle/job dispatch first, then translates RTVI
UI bus messages produced by a ``UIWorker`` (``BusUICommandMessage``
and the four ``BusUITask*`` lifecycle carriers) into the matching
RTVI frames and queues them downstream, where the ``RTVIObserver``
wraps them into typed ``ui-command`` / ``ui-task`` envelopes for the
client. Only the worker that owns the RTVI processor performs this
translation; other workers skip it.
"""
await super().on_bus_message(message)
if self._rtvi is None:
return
frame: Frame | None = None
if isinstance(message, BusUICommandMessage):
frame = RTVIUICommandFrame(
command=message.command_name,
payload=message.payload,
)
elif isinstance(message, BusUITaskGroupStartedMessage):
frame = RTVIUITaskFrame(
data=UITaskGroupStartedData(
task_id=message.task_id,
agents=list(message.agents or []),
label=message.label,
cancellable=message.cancellable,
at=message.at,
)
)
elif isinstance(message, BusUITaskUpdateMessage):
frame = RTVIUITaskFrame(
data=UITaskUpdateData(
task_id=message.task_id,
agent_name=message.agent_name,
data=message.data,
at=message.at,
)
)
elif isinstance(message, BusUITaskCompletedMessage):
frame = RTVIUITaskFrame(
data=UITaskCompletedData(
task_id=message.task_id,
agent_name=message.agent_name,
status=message.status,
response=message.response,
at=message.at,
)
)
elif isinstance(message, BusUITaskGroupCompletedMessage):
frame = RTVIUITaskFrame(
data=UITaskGroupCompletedData(
task_id=message.task_id,
at=message.at,
)
)
if frame is not None:
await self.queue_frame(frame)
async def _cancel(self, *, reason: str | None = None):
"""Internal cancellation logic for the pipeline worker.

249
tests/test_pipeline_worker_ui_bridge.py Normal file
View File

@@ -0,0 +1,249 @@
#
# Copyright (c) 2026, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Tests for the native RTVI⇄bus UI bridge built into PipelineWorker.
Inbound: typed RTVI UI messages from the client (fired via the RTVI
processor's ``on_ui_message`` event) are republished onto the bus as a
broadcast ``BusUIEventMessage``. Outbound: ``BusUICommandMessage`` and
the four ``BusUITask*`` lifecycle carriers are translated into the
matching RTVI frames and queued downstream. The bridge is active only
when RTVI is enabled.
"""
import asyncio
import unittest
from pipecat.bus.ui_messages import (
_UI_CANCEL_TASK_BUS_EVENT_NAME,
_UI_SNAPSHOT_BUS_EVENT_NAME,
BusUICommandMessage,
BusUIEventMessage,
BusUITaskCompletedMessage,
BusUITaskGroupCompletedMessage,
BusUITaskGroupStartedMessage,
BusUITaskUpdateMessage,
)
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.filters.identity_filter import IdentityFilter
from pipecat.processors.frame_processor import FrameDirection
from pipecat.processors.frameworks.rtvi.frames import RTVIUICommandFrame, RTVIUITaskFrame
from pipecat.processors.frameworks.rtvi.models import (
A11yNode,
A11ySnapshot,
UICancelTaskData,
UICancelTaskMessage,
UIEventData,
UIEventMessage,
UISnapshotData,
UISnapshotMessage,
)
def _make_root(*, enable_rtvi=True):
"""A PipelineWorker with bus + frame spies installed."""
worker = PipelineWorker(
Pipeline([IdentityFilter()]),
name="root",
enable_rtvi=enable_rtvi,
cancel_on_idle_timeout=False,
)
sent: list = []
frames: list = []
async def _record_bus(message):
sent.append(message)
async def _record_frame(frame, direction=FrameDirection.DOWNSTREAM):
frames.append(frame)
worker.send_bus_message = _record_bus # type: ignore[method-assign]
worker.queue_frame = _record_frame # type: ignore[method-assign]
return worker, sent, frames
async def _fire_ui_message(worker, message):
"""Fire the RTVI processor's ``on_ui_message`` event and drain it."""
await worker.rtvi._call_event_handler("on_ui_message", message)
tasks = [t for (_name, t) in list(worker.rtvi._event_tasks)]
if tasks:
await asyncio.gather(*tasks)
class TestUIBridgeInbound(unittest.IsolatedAsyncioTestCase):
async def test_republishes_ui_event_as_broadcast_bus_message(self):
worker, sent, _frames = _make_root()
await _fire_ui_message(
worker,
UIEventMessage(id="m1", data=UIEventData(event="nav_click", payload={"view": "home"})),
)
events = [m for m in sent if isinstance(m, BusUIEventMessage)]
self.assertEqual(len(events), 1)
self.assertEqual(events[0].source, "root")
self.assertIsNone(events[0].target)
self.assertEqual(events[0].event_name, "nav_click")
self.assertEqual(events[0].payload, {"view": "home"})
async def test_snapshot_message_routes_to_internal_event_name(self):
worker, sent, _frames = _make_root()
tree = A11ySnapshot(root=A11yNode(ref="root", role="document"), captured_at=1)
await _fire_ui_message(worker, UISnapshotMessage(id="m2", data=UISnapshotData(tree=tree)))
events = [m for m in sent if isinstance(m, BusUIEventMessage)]
self.assertEqual(events[0].event_name, _UI_SNAPSHOT_BUS_EVENT_NAME)
self.assertEqual(events[0].payload, tree.model_dump(exclude_none=True))
async def test_cancel_task_message_routes_to_internal_event_name(self):
worker, sent, _frames = _make_root()
await _fire_ui_message(
worker,
UICancelTaskMessage(id="m3", data=UICancelTaskData(task_id="t-1", reason="user")),
)
events = [m for m in sent if isinstance(m, BusUIEventMessage)]
self.assertEqual(events[0].event_name, _UI_CANCEL_TASK_BUS_EVENT_NAME)
self.assertEqual(events[0].payload, {"task_id": "t-1", "reason": "user"})
async def test_missing_payload_becomes_none(self):
worker, sent, _frames = _make_root()
await _fire_ui_message(worker, UIEventMessage(id="m1", data=UIEventData(event="hello")))
events = [m for m in sent if isinstance(m, BusUIEventMessage)]
self.assertEqual(events[0].event_name, "hello")
self.assertIsNone(events[0].payload)
async def test_unknown_message_type_is_ignored(self):
worker, sent, _frames = _make_root()
await _fire_ui_message(worker, object())
self.assertEqual([m for m in sent if isinstance(m, BusUIEventMessage)], [])
class TestUIBridgeOutbound(unittest.IsolatedAsyncioTestCase):
async def test_command_becomes_rtvi_ui_command_frame(self):
worker, _sent, frames = _make_root()
await worker.on_bus_message(
BusUICommandMessage(source="ui", target=None, command_name="toast", payload={"t": "Hi"})
)
ui_frames = [f for f in frames if isinstance(f, RTVIUICommandFrame)]
self.assertEqual(len(ui_frames), 1)
self.assertEqual(ui_frames[0].command, "toast")
self.assertEqual(ui_frames[0].payload, {"t": "Hi"})
async def test_group_started_envelope(self):
worker, _sent, frames = _make_root()
await worker.on_bus_message(
BusUITaskGroupStartedMessage(
source="ui",
target=None,
task_id="t1",
agents=["w1", "w2"],
label="Doing stuff",
cancellable=True,
at=1700,
)
)
frame = next(f for f in frames if isinstance(f, RTVIUITaskFrame))
self.assertEqual(frame.data.kind, "group_started")
self.assertEqual(frame.data.task_id, "t1")
self.assertEqual(frame.data.agents, ["w1", "w2"])
self.assertEqual(frame.data.label, "Doing stuff")
self.assertTrue(frame.data.cancellable)
self.assertEqual(frame.data.at, 1700)
async def test_task_update_envelope(self):
worker, _sent, frames = _make_root()
await worker.on_bus_message(
BusUITaskUpdateMessage(
source="ui",
target=None,
task_id="t1",
agent_name="w1",
data={"kind": "tool_call", "tool": "WebSearch"},
at=1701,
)
)
frame = next(f for f in frames if isinstance(f, RTVIUITaskFrame))
self.assertEqual(frame.data.kind, "task_update")
self.assertEqual(frame.data.task_id, "t1")
self.assertEqual(frame.data.agent_name, "w1")
self.assertEqual(frame.data.data, {"kind": "tool_call", "tool": "WebSearch"})
self.assertEqual(frame.data.at, 1701)
async def test_task_completed_envelope(self):
worker, _sent, frames = _make_root()
await worker.on_bus_message(
BusUITaskCompletedMessage(
source="ui",
target=None,
task_id="t1",
agent_name="w1",
status="completed",
response={"answer": 42},
at=1702,
)
)
frame = next(f for f in frames if isinstance(f, RTVIUITaskFrame))
self.assertEqual(frame.data.kind, "task_completed")
self.assertEqual(frame.data.task_id, "t1")
self.assertEqual(frame.data.agent_name, "w1")
self.assertEqual(frame.data.status, "completed")
self.assertEqual(frame.data.response, {"answer": 42})
self.assertEqual(frame.data.at, 1702)
async def test_group_completed_envelope(self):
worker, _sent, frames = _make_root()
await worker.on_bus_message(
BusUITaskGroupCompletedMessage(source="ui", target=None, task_id="t1", at=1703)
)
frame = next(f for f in frames if isinstance(f, RTVIUITaskFrame))
self.assertEqual(frame.data.kind, "group_completed")
self.assertEqual(frame.data.task_id, "t1")
self.assertEqual(frame.data.at, 1703)
async def test_non_ui_bus_message_queues_no_frame(self):
worker, _sent, frames = _make_root()
# A plain BusUIEventMessage (inbound carrier) is not an outbound
# command/task, so the outbound translation must ignore it.
await worker.on_bus_message(
BusUIEventMessage(source="x", target=None, event_name="e", payload={})
)
self.assertEqual(
[f for f in frames if isinstance(f, (RTVIUICommandFrame, RTVIUITaskFrame))], []
)
async def test_worker_without_rtvi_does_not_translate(self):
worker, _sent, frames = _make_root(enable_rtvi=False)
self.assertIsNone(worker._rtvi)
await worker.on_bus_message(
BusUICommandMessage(source="ui", target=None, command_name="toast", payload={})
)
self.assertEqual(frames, [])
if __name__ == "__main__":
unittest.main()