From 02667a7255f610d14775cb3fe79ec1d6592333b0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 21 May 2026 16:51:22 -0400 Subject: [PATCH] =?UTF-8?q?Add=20native=20RTVI=E2=87=84bus=20UI=20bridge?= =?UTF-8?q?=20to=20PipelineWorker?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/pipecat/bus/ui_messages.py | 178 +++++++++++++++++ src/pipecat/pipeline/worker.py | 111 +++++++++++ tests/test_pipeline_worker_ui_bridge.py | 249 ++++++++++++++++++++++++ 3 files changed, 538 insertions(+) create mode 100644 src/pipecat/bus/ui_messages.py create mode 100644 tests/test_pipeline_worker_ui_bridge.py diff --git a/src/pipecat/bus/ui_messages.py b/src/pipecat/bus/ui_messages.py new file mode 100644 index 000000000..59c4b6f00 --- /dev/null +++ b/src/pipecat/bus/ui_messages.py @@ -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 diff --git a/src/pipecat/pipeline/worker.py b/src/pipecat/pipeline/worker.py index 8466cbaf3..7989b0c19 100644 --- a/src/pipecat/pipeline/worker.py +++ b/src/pipecat/pipeline/worker.py @@ -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. diff --git a/tests/test_pipeline_worker_ui_bridge.py b/tests/test_pipeline_worker_ui_bridge.py new file mode 100644 index 000000000..74ab38157 --- /dev/null +++ b/tests/test_pipeline_worker_ui_bridge.py @@ -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()