diff --git a/changelog/xxxx.added.md b/changelog/xxxx.added.md
new file mode 100644
index 000000000..f0dd7fe15
--- /dev/null
+++ b/changelog/xxxx.added.md
@@ -0,0 +1,3 @@
+- Added `pipecat.workers.ui.UIWorker`, an `LLMContextWorker` that observes and drives a client GUI over the RTVI UI channel: it stores live accessibility snapshots, auto-injects `<ui_state>` at the start of each `respond` job, dispatches client events to `@on_ui_event` handlers, and sends UI commands (`scroll_to`, `highlight`, `select_text`, `click`, `set_input_value`) back to the client. The optional `ReplyToolMixin` exposes a bundled `reply` tool, and `user_job_group(...)` surfaces fan-out work to the client as cancellable task cards. A native RTVI⇄bus UI bridge is built into `PipelineWorker` (active whenever RTVI is enabled), so no decorator or manual wiring is needed: inbound UI messages are broadcast on the bus as `BusUIEventMessage`, and outbound `BusUICommandMessage` / `BusUITask*` carriers are translated into RTVI frames for the client.
+
+- `UIWorker` auto-injects the UI wire-format guide (`UI_STATE_PROMPT_GUIDE`) into its LLM's system instruction by default, via a `prompt_guide` parameter — pass your own string to override the guide, or `None` to disable. Apps no longer need to concatenate `UI_STATE_PROMPT_GUIDE` into the LLM's `system_instruction` by hand.
diff --git a/src/pipecat/workers/ui/__init__.py b/src/pipecat/workers/ui/__init__.py
new file mode 100644
index 000000000..03ae5e6c6
--- /dev/null
+++ b/src/pipecat/workers/ui/__init__.py
@@ -0,0 +1,49 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""UI worker: LLM worker that observes and drives a GUI app.
+
+Composes the RTVI wire protocol for client UI events, accessibility-tree
+snapshots, and server-emitted UI commands (in
+``pipecat.processors.frameworks.rtvi.models``) with an opt-in
+``ReplyToolMixin`` exposing the canonical bundled reply tool
+(``answer`` + optional ``scroll_to`` / ``highlight`` / ...).
+
+The RTVI⇄bus bridge that connects a ``UIWorker`` to the client is built
+into ``PipelineWorker`` and is active whenever RTVI is enabled — there
+is no decorator or bridge to wire up.
+"""
+
+from pipecat.bus.ui_messages import (
+    BusUICommandMessage,
+    BusUIEventMessage,
+    BusUITaskCompletedMessage,
+    BusUITaskGroupCompletedMessage,
+    BusUITaskGroupStartedMessage,
+    BusUITaskUpdateMessage,
+)
+from pipecat.workers.ui.ui_event_decorator import on_ui_event
+from pipecat.workers.ui.ui_prompts import UI_STATE_PROMPT_GUIDE
+from pipecat.workers.ui.ui_tools import ReplyToolMixin
+from pipecat.workers.ui.ui_worker import UIWorker
+
+# Built-in UI command payload models (Toast, Navigate, ScrollTo,
+# Highlight, Focus, Click, SetInputValue, SelectText) live in
+# ``pipecat.processors.frameworks.rtvi.models``. Import them from there
+# directly.
+
+__all__ = [
+    "BusUICommandMessage",
+    "BusUIEventMessage",
+    "BusUITaskCompletedMessage",
+    "BusUITaskGroupCompletedMessage",
+    "BusUITaskGroupStartedMessage",
+    "BusUITaskUpdateMessage",
+    "ReplyToolMixin",
+    "UIWorker",
+    "UI_STATE_PROMPT_GUIDE",
+    "on_ui_event",
+]
diff --git a/src/pipecat/workers/ui/ui_event_decorator.py b/src/pipecat/workers/ui/ui_event_decorator.py
new file mode 100644
index 000000000..a4d5c0b22
--- /dev/null
+++ b/src/pipecat/workers/ui/ui_event_decorator.py
@@ -0,0 +1,67 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Decorator for marking worker methods as UI event handlers."""
+
+
+def on_ui_event(name: str):
+    """Mark a worker method as a handler for a named UI event.
+
+    On ``UIWorker`` subclasses, decorated methods are automatically
+    dispatched when a ``BusUIEventMessage`` with a matching ``name``
+    arrives.
+
+    Example::
+
+        class MyUIWorker(UIWorker):
+            @on_ui_event("nav_click")
+            async def on_nav(self, message):
+                view = message.payload.get("view")
+                ...
+
+    Args:
+        name: The UI event name to match.
+    """
+
+    def decorator(fn):
+        fn.is_ui_event_handler = True
+        fn.ui_event_name = name
+        return fn
+
+    return decorator
+
+
+def _collect_ui_event_handlers(obj) -> dict:
+    """Collect all ``@on_ui_event`` decorated bound methods from an object.
+
+    Walks the MRO so that overridden methods in subclasses take
+    precedence over base-class definitions.
+
+    Returns:
+        A dict mapping event name to the bound method.
+
+    Raises:
+        ValueError: If two handlers share the same event name on the
+            same subclass level.
+    """
+    seen: set[str] = set()
+    handlers: dict[str, object] = {}
+    source_names: dict[str, str] = {}  # event name -> defining method name, for errors
+    for cls in type(obj).__mro__:
+        for attr_name, val in cls.__dict__.items():
+            if attr_name in seen:
+                continue
+            seen.add(attr_name)
+            if callable(val) and getattr(val, "is_ui_event_handler", False):
+                event_name = val.ui_event_name
+                if event_name in handlers:
+                    raise ValueError(
+                        f"Duplicate @on_ui_event handler for '{event_name}': "
+                        f"'{attr_name}' conflicts with '{source_names[event_name]}'"
+                    )
+                handlers[event_name] = getattr(obj, attr_name)
+                source_names[event_name] = attr_name
+    return handlers
diff --git a/src/pipecat/workers/ui/ui_job_context.py b/src/pipecat/workers/ui/ui_job_context.py
new file mode 100644
index 000000000..4b72df2ec
--- /dev/null
+++ b/src/pipecat/workers/ui/ui_job_context.py
@@ -0,0 +1,147 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User-facing job group context.
+
+Wraps ``JobGroupContext`` so the work it dispatches is also surfaced
+to the UI client through the UI Worker protocol. Apps reach this via
+``UIWorker.user_job_group(...)`` rather than constructing it directly.
+"""
+
+from __future__ import annotations
+
+import time
+from typing import TYPE_CHECKING
+
+from pipecat.bus.ui_messages import (
+    BusUITaskGroupCompletedMessage,
+    BusUITaskGroupStartedMessage,
+)
+from pipecat.pipeline.job_context import JobGroupContext
+
+if TYPE_CHECKING:
+    from pipecat.workers.ui.ui_worker import UIWorker
+
+
+class UserJobGroupContext(JobGroupContext):
+    """Job group whose lifecycle is forwarded to the UI client.
+
+    Behaves exactly like ``JobGroupContext`` for the dispatching code.
+    Additionally, on enter the context registers the group with its
+    parent ``UIWorker`` and publishes a ``BusUITaskGroupStartedMessage``.
+    The worker forwards any subsequent ``BusJobUpdateMessage`` /
+    ``BusJobResponseMessage`` whose ``job_id`` matches a registered
+    group as ``BusUITaskUpdateMessage`` / ``BusUITaskCompletedMessage``.
+    On exit the context publishes ``BusUITaskGroupCompletedMessage`` and
+    deregisters.
+
+    Workers don't need to know about the UI surface: any
+    ``send_job_update`` they emit against the group's ``job_id`` is
+    forwarded automatically.
+
+    Example::
+
+        async with self.user_job_group(
+            "researcher_a", "researcher_b",
+            payload={"query": query},
+            label=f"Research: {query}",
+            cancellable=True,
+        ) as tg:
+            async for event in tg:
+                ...
+            results = tg.responses
+    """
+
+    def __init__(
+        self,
+        worker: UIWorker,
+        worker_names: tuple[str, ...],
+        *,
+        name: str | None = None,
+        payload: dict | None = None,
+        timeout: float | None = None,
+        cancel_on_error: bool = True,
+        label: str | None = None,
+        cancellable: bool = True,
+    ):
+        """Initialize the UserJobGroupContext.
+
+        Args:
+            worker: The parent ``UIWorker`` that owns this job group.
+            worker_names: Names of the workers to send the job to.
+            name: Optional job name for routing to named ``@job``
+                handlers on the workers.
+            payload: Optional structured data describing the work.
+            timeout: Optional timeout in seconds covering both the
+                ready-wait and job execution.
+            cancel_on_error: Whether to cancel the group if a worker
+                errors. Defaults to True.
+            label: Optional human-readable label surfaced to the
+                client (e.g. ``"Research: Radiohead"``). The client UI
+                uses it to title the in-flight task card.
+            cancellable: Whether the client may request cancellation
+                of this group via the reserved ``__cancel_task`` event.
+                Defaults to True.
+        """
+        super().__init__(
+            worker,
+            worker_names,
+            name=name,
+            payload=payload,
+            timeout=timeout,
+            cancel_on_error=cancel_on_error,
+        )
+        self._ui_worker = worker
+        self._label = label
+        self._cancellable = cancellable
+
+    @property
+    def label(self) -> str | None:
+        """The group's human-readable label, if any."""
+        return self._label
+
+    @property
+    def cancellable(self) -> bool:
+        """Whether the client may request cancellation."""
+        return self._cancellable
+
+    async def __aenter__(self) -> UserJobGroupContext:
+        await super().__aenter__()
+        job_id = self.job_id
+        self._ui_worker._register_user_job_group(
+            job_id=job_id,
+            worker_names=list(self._worker_names),
+            label=self._label,
+            cancellable=self._cancellable,
+        )
+        await self._ui_worker.send_bus_message(
+            BusUITaskGroupStartedMessage(
+                source=self._ui_worker.name,
+                target=None,
+                task_id=job_id,
+                agents=list(self._worker_names),
+                label=self._label,
+                cancellable=self._cancellable,
+                at=int(time.time() * 1000),
+            )
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb) -> bool:
+        job_id = self._group.job_id if self._group else None
+        try:
+            return await super().__aexit__(exc_type, exc_val, exc_tb)
+        finally:
+            if job_id:
+                self._ui_worker._unregister_user_job_group(job_id)
+                await self._ui_worker.send_bus_message(
+                    BusUITaskGroupCompletedMessage(
+                        source=self._ui_worker.name,
+                        target=None,
+                        task_id=job_id,
+                        at=int(time.time() * 1000),
+                    )
+                )
diff --git a/src/pipecat/workers/ui/ui_prompts.py b/src/pipecat/workers/ui/ui_prompts.py
new file mode 100644
index 000000000..961c3ed05
--- /dev/null
+++ b/src/pipecat/workers/ui/ui_prompts.py
@@ -0,0 +1,64 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Canonical prompt fragments describing the UI worker wire format.
+
+Apps concatenate these constants into their system prompt so the
+LLM understands the ``<ui_state>`` and ``<ui_event>`` developer
+messages the SDK injects on its behalf.
+
+Example::
+
+    system_prompt = f'''
+    You are a voice-driven music player agent.
+    ...app-specific tool and behavior instructions...
+
+    {UI_STATE_PROMPT_GUIDE}
+    '''
+
+The SDK updates the guide alongside the wire format, so apps get
+new tags and semantics automatically on the next release.
+"""
+
+UI_STATE_PROMPT_GUIDE: str = """\
+## UI context
+
+Your developer context includes two kinds of SDK-managed messages:
+
+- ``<ui_event name="..." >payload</ui_event>``: an event the user just \
+triggered on the client (click, tab switch, navigation, etc.). The \
+payload is JSON for that event.
+- ``<ui_state>...</ui_state>``: an accessibility snapshot of the \
+current screen, injected at the start of every task request. \
+Indented tree in Playwright-MCP style. Each line is \
+``- role "name" [state] [ref=eN]`` with children nested one level \
+deeper.
+
+State tags include ``[focused]``, ``[selected]``, ``[disabled]``, and \
+``[offscreen]``. A node tagged ``[offscreen]`` exists on the page \
+but is not currently in the user's viewport; only visible \
+(non-offscreen) nodes count for position-based references.
+
+Grids carry a ``[cols=N]`` tag. Their cells are listed in reading \
+order (left-to-right, top-to-bottom); with N columns, cell K sits \
+at row ``ceil(K/N)``, column ``((K-1) mod N) + 1``. Example with \
+``[cols=8]`` and 16 children: "top right" is cell 8, "bottom left" \
+is cell 9.
+
+Resolve position references ("top right", "the first one", "the \
+third new release") against the most recent ``<ui_state>`` tree. \
+Sibling order matches reading order on screen (top-to-bottom, \
+left-to-right within each region).
+
+When the user has text selected on the page, the snapshot ends with \
+a ``<selection ref="eN">selected text</selection>`` block inside \
+``<ui_state>``. Treat the selection as the deictic referent for \
+"this", "that", "what I selected", and similar phrases. The ``ref`` \
+identifies the closest enclosing element that has a ref in the tree; \
+the inner text is the actual selected content (truncated if very \
+long). Text inside ``<input>`` or ``<textarea>`` selections is \
+faithful to ``selectionStart``/``selectionEnd`` on the element.\
+"""
diff --git a/src/pipecat/workers/ui/ui_tools.py b/src/pipecat/workers/ui/ui_tools.py
new file mode 100644
index 000000000..deeb32b7a
--- /dev/null
+++ b/src/pipecat/workers/ui/ui_tools.py
@@ -0,0 +1,155 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Opt-in tool mixin for ``UIWorker``.
+
+Ships ``ReplyToolMixin``: a single ``reply(answer, scroll_to,
+highlight, select_text, fills, click)`` LLM tool that bundles a
+required spoken answer with the full set of standard UI actions.
+One tool call per turn; the model cannot drop the terminator because
+``answer`` is a required argument that the API schema enforces.
+
+The bundled mixin covers the canonical app shapes (pointing,
+reading, form-fill) and any blend of them. Apps don't need to pick
+a mode up front; the LLM uses whichever fields make sense per turn,
+leaving the rest as ``null``.
+
+Apps that want a tighter schema (only the fields they use, or
+app-specific commands like ``play_song``) hand-roll their own
+``@tool reply`` on the ``UIWorker`` subclass directly. The helper
+methods on ``UIWorker`` (``scroll_to``, ``highlight``,
+``select_text``, ``click``, ``set_input_value``) plus
+``send_command`` and the standard payload dataclasses cover the
+building blocks for custom replies.
+"""
+
+from loguru import logger
+
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.workers.llm.tool_decorator import tool
+
+
+class ReplyToolMixin:
+    """Expose a ``reply`` tool covering the full standard action set.
+
+    Single bundled LLM tool with a required spoken ``answer`` plus
+    optional visual and state-changing actions. One tool call per
+    turn, no chaining; the required ``answer`` argument is enforced
+    by the API schema so the model cannot omit the terminator.
+
+    Compose alongside ``UIWorker``::
+
+        class MyUIWorker(ReplyToolMixin, UIWorker):
+            ...
+
+    Covers pointing apps (``scroll_to`` + ``highlight``), reading
+    apps (``scroll_to`` + ``select_text``), form apps (``fills`` +
+    ``click``), and any blend (e.g. a document review with
+    selection-based deixis AND voice-driven note-taking). The LLM
+    uses whichever fields fit the user's request per turn; unused
+    fields stay ``null`` and don't affect behavior.
+
+    Apps that want a minimal schema (only the fields actually used,
+    or app-specific commands) write their own ``@tool reply`` on the
+    ``UIWorker`` subclass directly. Use the helper methods on
+    ``UIWorker`` plus ``send_command`` to dispatch the underlying UI
+    commands.
+
+    The host class must provide ``scroll_to``, ``highlight``,
+    ``select_text``, ``click``, ``set_input_value``, and
+    ``respond_to_job`` (``UIWorker`` does) and must be the target of
+    ``@tool`` discovery on the LLM pipeline.
+    """
+
+    @tool
+    async def reply(
+        self,
+        params: FunctionCallParams,
+        answer: str,
+        scroll_to: str | None = None,
+        highlight: list[str] | None = None,
+        select_text: str | None = None,
+        fills: list[dict] | None = None,
+        click: list[str] | None = None,
+    ):
+        """Reply to the user. Optionally point at content and act on inputs.
+
+        Always called exactly once per turn. ``answer`` is required;
+        the action fields are optional and may be combined.
+
+        Visual / pointing actions (draw the user's attention):
+
+        - ``scroll_to`` brings an element into view (single ref).
+        - ``highlight`` flashes elements briefly (list of refs).
+          Best for short emphasis like a button or a fact.
+        - ``select_text`` puts the page's text selection on an
+          element (single ref). Best for "this paragraph" / "the
+          section about X" so the user sees exactly what was meant.
+          Persists until the user clicks elsewhere.
+
+        State-changing actions (modify form / app state):
+
+        - ``fills`` writes values into inputs (list of
+          ``{"ref", "value"}`` objects, multi-fill in one turn).
+        - ``click`` clicks elements (list of refs in order). Use for
+          checkboxes, radios, submit buttons.
+
+        Order of dispatch within a turn: ``scroll_to``, then
+        ``highlight``, then ``select_text``, then ``fills``, then
+        ``click``, then speak the answer.
+
+        Args:
+            params: Framework-provided tool invocation context.
+            answer: The spoken reply in plain language. One short
+                sentence. No markdown, no symbols.
+            scroll_to: Optional snapshot ref. Scrolls the element
+                into view before speaking.
+            highlight: Optional list of snapshot refs. Visually
+                pulses each element.
+            select_text: Optional snapshot ref. Places the page's
+                text selection on that element.
+            fills: Optional list of ``{"ref": "eN", "value": "..."}``
+                objects. Writes each value into the input at ``ref``.
+            click: Optional list of snapshot refs to click in order.
+        """
+        preview = (answer or "").strip()
+        if len(preview) > 80:
+            preview = preview[:80] + "…"
+        logger.info(
+            f"{self}: reply(answer={preview!r}, scroll_to={scroll_to!r}, "
+            f"highlight={highlight!r}, select_text={select_text!r}, "
+            f"fills={fills!r}, click={click!r})"
+        )
+        # Defensive guards on the list arguments: an LLM that emits a
+        # malformed entry (None, a bare string, etc.) would crash the
+        # tool body before respond_to_job fires, leaving the
+        # single-flight lock held until the requester's timeout cancels
+        # us. Skip non-conforming entries instead.
+        if scroll_to:
+            await self.scroll_to(scroll_to)  # type: ignore[attr-defined]
+        if highlight:
+            for ref in highlight:
+                if not isinstance(ref, str):
+                    continue
+                await self.highlight(ref)  # type: ignore[attr-defined]
+        if select_text:
+            await self.select_text(select_text)  # type: ignore[attr-defined]
+        if fills:
+            for entry in fills:
+                if not isinstance(entry, dict):
+                    continue
+                ref = entry.get("ref")
+                value = entry.get("value")
+                if not isinstance(ref, str) or value is None:
+                    continue
+                await self.set_input_value(ref, str(value))  # type: ignore[attr-defined]
+        if click:
+            for ref in click:
+                if not isinstance(ref, str):
+                    continue
+                await self.click(ref)  # type: ignore[attr-defined]
+        await self.respond_to_job(speak=answer)  # type: ignore[attr-defined]
+        await params.result_callback(None)
diff --git a/src/pipecat/workers/ui/ui_worker.py b/src/pipecat/workers/ui/ui_worker.py
new file mode 100644
index 000000000..3f45e533d
--- /dev/null
+++ b/src/pipecat/workers/ui/ui_worker.py
@@ -0,0 +1,1151 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""UIWorker: LLM worker that dispatches UI events from the client."""
+
+import asyncio
+import json
+import time
+from dataclasses import asdict, dataclass, is_dataclass
+from typing import Any
+
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.bus.messages import (
+    BusJobRequestMessage,
+    BusJobResponseMessage,
+    BusJobResponseUrgentMessage,
+    BusJobUpdateMessage,
+    BusJobUpdateUrgentMessage,
+    BusMessage,
+)
+from pipecat.bus.ui_messages import (
+    _UI_CANCEL_TASK_BUS_EVENT_NAME,
+    _UI_SNAPSHOT_BUS_EVENT_NAME,
+    BusUICommandMessage,
+    BusUIEventMessage,
+    BusUITaskCompletedMessage,
+    BusUITaskUpdateMessage,
+)
+from pipecat.frames.frames import LLMMessagesAppendFrame, LLMMessagesUpdateFrame
+from pipecat.pipeline.job_context import JobGroupError, JobStatus
+from pipecat.pipeline.job_decorator import job
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMAssistantAggregatorParams,
+    LLMUserAggregatorParams,
+)
+from pipecat.processors.frameworks.rtvi.models import (
+    Click,
+    Highlight,
+    ScrollTo,
+    SelectText,
+    SetInputValue,
+)
+from pipecat.services.llm_service import LLMService
+from pipecat.workers.llm.llm_context_worker import LLMContextWorker
+from pipecat.workers.ui.ui_event_decorator import _collect_ui_event_handlers
+from pipecat.workers.ui.ui_job_context import UserJobGroupContext
+from pipecat.workers.ui.ui_prompts import UI_STATE_PROMPT_GUIDE
+
+
+@dataclass
+class _UserJobGroupRegistration:
+    """Metadata kept on a UIWorker for each in-flight user job group.
+
+    The registry is the lookup the worker uses to decide which bus
+    job messages to forward to the client and whether a client
+    ``__cancel_task`` event should be honored.
+    """
+
+    worker_names: list[str]
+    label: str | None
+    cancellable: bool
+
+
+class UIWorker(LLMContextWorker):
+    """LLM worker that dispatches UI events from the client.
+
+    Receives ``BusUIEventMessage`` (republished by the native UI bridge
+    in ``PipelineWorker``) and dispatches each one to the matching
+    ``@on_ui_event(name)`` handler. By default every event is also
+    appended to the LLM context as
+    ``<ui_event name="...">payload</ui_event>`` so the worker can reason
+    about what the user just did on the next inference.
+
+    ## Canonical pattern
+
+    A ``UIWorker`` is the delegate side of a voice ↔ UI split: a
+    bridged ``LLMWorker`` (the voice layer) receives the user's
+    transcript and delegates UI-relevant work to this worker via
+    ``self.job("ui_worker_name", name="respond", payload={"query": text})``.
+    The built-in ``respond`` job runs: ``<ui_state>`` is auto-injected,
+    the LLM picks a tool, and the job completes with a spoken reply
+    the voice worker hands to TTS.
+
+    A working subclass needs two things: an LLM and a ``@tool`` that
+    calls ``respond_to_job`` to finish the turn. The built-in
+    ``@job(name="respond", sequential=True)`` handler drives the LLM
+    for you (see ``respond_with_llm``); the pipeline, activation,
+    snapshot injection, and tool registration come from the base
+    classes. Override ``render_query`` to read a non-default payload
+    shape.
+
+    Example::
+
+        class MyUIWorker(UIWorker):
+            @on_ui_event("nav_click")
+            async def on_nav(self, message):
+                view = message.payload.get("view")
+                ...
+
+            @tool
+            async def answer(self, params, text: str):
+                await self.respond_to_job(speak=text)
+                await params.result_callback(None)
+
+        worker = MyUIWorker("ui", llm=OpenAILLMService(api_key="..."))
+
+    ## Activation default
+
+    Unlike ``LLMWorker`` and ``LLMContextWorker`` (which default to
+    ``active=False`` because voice handoff patterns require
+    parent-driven activation), ``UIWorker`` defaults to ``active=True``.
+    UIWorkers are typically always-on delegates with no equivalent of
+    "the user is now talking to the UIWorker" — the worker activates as
+    soon as its pipeline starts, registers its tools, and stays online
+    to receive jobs. Pass ``active=False`` explicitly if you have a
+    handoff use case.
+
+    ## Single-flight job semantics
+
+    A ``UIWorker`` owns a single ``LLMContext`` and one running pipeline
+    (``LLMContextWorker`` builds ``[user_aggregator, llm,
+    assistant_aggregator]``). Every LLM turn reads and mutates that one
+    context, so two ``respond`` turns cannot run at once without
+    interleaving their ``<ui_state>`` injections and corrupting history.
+    Serialization comes from the framework: the built-in handler is
+    ``@job(name="respond", sequential=True)``, so same-name requests
+    queue and run one at a time in arrival order (e.g. when the voice
+    LLM issues parallel tool calls, which pipecat runs concurrently by
+    default).
+
+    For this to hold across the full turn, ``respond_with_llm`` *spans
+    the work*: it drives the LLM and then blocks on an internal future
+    until a ``@tool`` calls ``respond_to_job``. The framework holds the
+    per-name lock for the handler's whole lifetime, so the next
+    ``respond`` request waits until the current turn is fully answered.
+    A tool that never calls ``respond_to_job`` leaves the handler
+    blocked and subsequent ``respond`` requests queued; that hang is an
+    intentional, fast-surfacing signal of a missing terminator, and the
+    requester's job timeout provides error-path liveness. Don't add a
+    watchdog.
+
+    Concurrency is per job name, not per worker. A *different*-named
+    ``@job`` handler runs concurrently with ``respond`` — fine for work
+    that does NOT touch the LLM context (e.g. a pure data fetch or UI
+    action that replies via ``send_job_response`` directly), but two
+    LLM-driving job types on one worker would race on the shared
+    context. For genuinely parallel LLM work, run multiple UIWorkers.
+
+    ## Visibility convention
+
+    When the client captures snapshots with ``trackViewport`` enabled
+    (the default), every node in the accessibility tree whose bounding
+    rect sits fully outside the viewport carries ``"offscreen"`` in its
+    ``state`` list. The rendered ``<ui_state>`` surfaces this as
+    ``[offscreen]``. ``visible_nodes()`` returns the filtered subset
+    the user is actually looking at right now. Workers should treat
+    ``[offscreen]`` nodes as on-the-page-but-not-in-view, typically
+    issuing a ``scroll_to`` command before acting on them.
+    """
+
+    #: Job name of the built-in LLM ``respond`` handler. Requesters
+    #: dispatch with ``self.job(worker, name=UIWorker.DEFAULT_JOB_NAME,
+    #: payload={"query": ...})``. Kept in sync with the literal in the
+    #: ``@job(name="respond", ...)`` decorator below.
+    DEFAULT_JOB_NAME = "respond"
+
+    def __init__(
+        self,
+        name: str,
+        *,
+        llm: LLMService[Any],
+        active: bool = True,
+        bridged: tuple[str, ...] | None = None,
+        defer_tool_frames: bool = True,
+        context: LLMContext | None = None,
+        user_params: LLMUserAggregatorParams | None = None,
+        assistant_params: LLMAssistantAggregatorParams | None = None,
+        inject_events: bool = True,
+        auto_inject_ui_state: bool = True,
+        keep_history: bool = False,
+        prompt_guide: str | None = UI_STATE_PROMPT_GUIDE,
+    ):
+        """Initialize the UIWorker.
+
+        Args:
+            name: Unique name for this worker.
+            llm: The LLM service.
+            active: Whether the worker starts active. Defaults to
+                ``True`` for ``UIWorker`` (vs. ``False`` on
+                ``LLMWorker`` / ``LLMContextWorker``) because the
+                canonical UIWorker role is an always-on delegate that
+                should self-activate as soon as its pipeline starts.
+                Pass ``active=False`` only if you have a handoff use
+                case.
+            bridged: Bridge configuration. See ``PipelineWorker`` for
+                details.
+            defer_tool_frames: Forwarded to ``LLMContextWorker``. See
+                ``LLMWorker`` for details.
+            context: Optional pre-built ``LLMContext``. Forwarded to
+                ``LLMContextWorker``. Note that any messages seeded
+                here are part of the mutable job history and are
+                cleared before each job when ``keep_history=False``
+                (the default), since the reset replaces the entire
+                message list. For durable UI / app instructions,
+                pass them via the LLM service's ``system_instruction``
+                instead, where they live outside the context message
+                list and survive resets.
+            user_params: Optional user-aggregator parameters.
+                Forwarded to ``LLMContextWorker``.
+            assistant_params: Optional assistant-aggregator parameters.
+                Forwarded to ``LLMContextWorker``.
+            prompt_guide: Wire-format guide appended to the LLM's
+                ``system_instruction`` so it can parse the ``<ui_state>``
+                and ``<ui_event>`` developer messages this worker injects.
+                Defaults to ``UI_STATE_PROMPT_GUIDE``; pass your own string
+                to override it, or ``None`` to disable. It is appended
+                *after* the LLM's existing system instruction (so put your
+                app prompt there as usual) and, living in
+                ``system_instruction`` rather than the message list, it
+                survives the per-job context reset.
+            inject_events: Whether to auto-append each UI event to the
+                LLM context as a ``<ui_event>`` developer message.
+                Defaults to True. Override ``render_ui_event`` to change
+                the injected content, or set this to False to disable.
+            auto_inject_ui_state: When True (the default), the latest
+                ``<ui_state>`` snapshot is appended to the LLM context
+                at the start of every job request, so the worker always
+                reasons over the current screen. Set to False if you
+                want to call ``inject_ui_state()`` yourself.
+            keep_history: When False (the default), the LLM context is
+                cleared at the start of every job. Each job starts
+                from an empty messages list, the current ``<ui_state>``
+                is injected, and the user's query follows. Best for
+                the canonical stateless-delegate role where the voice
+                worker owns dialog state and the UI worker's job is
+                "given the current screen, do something." When True,
+                the conversation history accumulates across jobs
+                (queries, prior ``<ui_state>`` blocks, tool calls,
+                responses) so the LLM can reason over multi-turn
+                references like "show me the next one" or "tell me
+                about the Pro version of that." History accumulation
+                grows token usage and can confuse smaller models when
+                multiple ``<ui_state>`` snapshots are present at once;
+                opt in only when the dialog continuity is worth the
+                cost. Apps in ``keep_history=True`` mode that
+                occasionally want to clear can call
+                ``await self.reset_context()``.
+
+                Note: in ``keep_history=False`` mode any messages
+                pre-seeded via ``context=`` are also cleared on the
+                first job, since the reset replaces the entire
+                message list. Durable UI / app instructions belong
+                in the LLM service's ``system_instruction`` setting
+                (which is sent at API call time and is not touched
+                by the context reset). The UI wire-format guide is
+                appended there automatically (see ``prompt_guide``), so
+                keep your app's durable instructions in
+                ``system_instruction`` too. Use ``keep_history=True`` if
+                the seeded messages genuinely need to live in the
+                conversation history.
+
+        Raises:
+            ValueError: If ``bridged`` is set together with the default
+                ``auto_inject_ui_state=True``. The two are
+                incompatible: auto-injection fires at the start of the
+                ``respond`` job, but a bridged ``UIWorker`` receives
+                user voice frames through the bridge instead of job
+                messages, so the snapshot would never reach the LLM
+                context. The canonical pattern is a non-bridged
+                ``UIWorker`` that receives delegated jobs from a
+                separate voice ``LLMWorker``. If you really want a
+                bridged ``UIWorker`` (advanced cases), pass
+                ``auto_inject_ui_state=False`` explicitly and call
+                ``inject_ui_state()`` yourself.
+        """
+        if bridged is not None and auto_inject_ui_state:
+            raise ValueError(
+                f"UIWorker '{name}': bridged + auto_inject_ui_state=True is "
+                "incompatible. Auto-injection fires at the start of the respond "
+                "job, but a bridged UIWorker receives frames through the bridge — the "
+                "snapshot would never land in the LLM context and the worker "
+                "would silently hallucinate. Use the canonical pattern "
+                "(non-bridged UIWorker receiving jobs from a separate "
+                "LLMWorker) or pass auto_inject_ui_state=False if you really "
+                "want a bridged UIWorker and will manage injection manually."
+            )
+        super().__init__(
+            name,
+            llm=llm,
+            active=active,
+            bridged=bridged,
+            defer_tool_frames=defer_tool_frames,
+            context=context,
+            user_params=user_params,
+            assistant_params=assistant_params,
+        )
+        # Auto-append the UI wire-format guide to the LLM's system
+        # instruction so the author doesn't have to concatenate it manually
+        # and it survives the per-job context reset. Pass a string to
+        # ``prompt_guide`` to override the text, or ``None`` to disable.
+        if prompt_guide:
+            self.llm.append_system_instruction(prompt_guide)
+        self._inject_events = inject_events
+        self._auto_inject_ui_state = auto_inject_ui_state
+        self._keep_history = keep_history
+        self._ui_event_handlers = _collect_ui_event_handlers(self)
+        # Latest accessibility snapshot received from the client. Updated
+        # in ``on_bus_message`` when a ``__ui_snapshot`` event arrives.
+        # Rendered into LLM context via ``inject_ui_state``.
+        self._latest_snapshot: dict[str, Any] | None = None
+        # Job currently being processed by this worker. Set in
+        # ``respond_with_llm``, cleared when the job completes. Lets
+        # ``@tool`` methods (and the mixin tools) close out the job
+        # without having to thread the job id through every call.
+        self._current_job: BusJobRequestMessage | None = None
+        # Resolved by ``respond_to_job`` to hand the result back to the
+        # in-flight ``respond_with_llm`` handler, which then sends the
+        # job response. See the "Single-flight job semantics" section
+        # in the class docstring.
+        self._pending: asyncio.Future | None = None
+        # Registry of in-flight user job groups dispatched by this
+        # worker (see ``user_job_group``). Keyed by ``job_id``.
+        # ``on_bus_message`` consults this to decide which job
+        # update / response messages should be forwarded to the
+        # client as ``ui-task`` envelopes.
+        self._user_job_groups: dict[str, _UserJobGroupRegistration] = {}
+
+    async def send_command(self, name: str, payload: Any = None) -> None:
+        """Send a named UI command to the client.
+
+        Publishes a ``BusUICommandMessage`` which the native UI bridge
+        in ``PipelineWorker`` translates into an ``RTVIUICommandFrame``
+        on the pipeline. Client-side handlers subscribed to
+        ``RTVIEvent.UICommand`` (or React's ``useUICommandHandler``)
+        dispatch on the command name.
+
+        Args:
+            name: App-defined command name (e.g. ``"toast"``,
+                ``"navigate"``, or any app-specific name).
+            payload: One of:
+
+                - A pydantic ``BaseModel`` instance (including the
+                  built-in command models in
+                  ``pipecat.processors.frameworks.rtvi.models``).
+                  Converted to a plain dict with ``model_dump()``.
+                - A dataclass instance. Converted to a plain dict with
+                  ``dataclasses.asdict``.
+                - A ``dict`` forwarded as-is.
+                - ``None``, forwarded as an empty dict.
+        """
+        if payload is None:
+            serialized: Any = {}
+        elif isinstance(payload, BaseModel):
+            serialized = payload.model_dump()
+        elif is_dataclass(payload) and not isinstance(payload, type):
+            serialized = asdict(payload)
+        else:
+            serialized = payload
+
+        await self.send_bus_message(
+            BusUICommandMessage(
+                source=self.name,
+                target=None,
+                command_name=name,
+                payload=serialized,
+            )
+        )
+
+    async def scroll_to(self, ref: str) -> None:
+        """Dispatch a standard ``scroll_to`` UI command for the given ref.
+
+        Convenience wrapper around ``send_command("scroll_to",
+        ScrollTo(ref=ref))``. Useful when composing custom ``@tool``
+        methods that bundle visual actions with a spoken reply (see
+        ``ReplyToolMixin`` for the canonical shape).
+
+        Note this is a regular instance method, not an LLM tool.
+        Subclasses that want to expose it to the LLM should write
+        their own ``@tool`` method that calls
+        ``await super().scroll_to(ref)``.
+
+        Args:
+            ref: Snapshot ref like ``"e42"`` from the most recent
+                ``<ui_state>``.
+        """
+        await self.send_command("scroll_to", ScrollTo(ref=ref))
+
+    async def highlight(self, ref: str) -> None:
+        """Dispatch a standard ``highlight`` UI command for the given ref.
+
+        Convenience wrapper around ``send_command("highlight",
+        Highlight(ref=ref))``. Same pattern as ``scroll_to``: a plain
+        instance method, not an LLM tool. Compose into a custom
+        ``@tool`` body, or use ``ReplyToolMixin`` for the standard
+        shape.
+
+        Args:
+            ref: Snapshot ref like ``"e42"`` from the most recent
+                ``<ui_state>``.
+        """
+        await self.send_command("highlight", Highlight(ref=ref))
+
+    async def select_text(
+        self,
+        ref: str,
+        *,
+        start_offset: int | None = None,
+        end_offset: int | None = None,
+    ) -> None:
+        """Dispatch a standard ``select_text`` UI command for the given ref.
+
+        Convenience wrapper around ``send_command("select_text",
+        SelectText(...))``. Same pattern as ``scroll_to`` and
+        ``highlight``: a plain instance method, not an LLM tool.
+        Compose into a custom ``@tool`` body for apps that want to
+        point at content via the page's text selection (deixis).
+
+        With ``start_offset`` / ``end_offset`` set, the client's
+        standard handler selects a sub-range of the element's text.
+        With both ``None`` (the default), the entire element's
+        contents are selected. The offsets are character offsets over
+        the element's concatenated ``textContent``.
+
+        Args:
+            ref: Snapshot ref like ``"e42"`` from the most recent
+                ``<ui_state>``.
+            start_offset: Optional character offset where the
+                selection should start. Pass with ``end_offset`` to
+                select a sub-range.
+            end_offset: Optional character offset where the selection
+                should end (exclusive).
+        """
+        await self.send_command(
+            "select_text",
+            SelectText(ref=ref, start_offset=start_offset, end_offset=end_offset),
+        )
+
+    async def click(self, ref: str) -> None:
+        """Dispatch a standard ``click`` UI command for the given ref.
+
+        Convenience wrapper around ``send_command("click",
+        Click(ref=ref))``. Same pattern as the other helpers: a plain
+        instance method, not an LLM tool. Use inside custom ``@tool``
+        bodies for state-changing form actions like checkboxes,
+        radio buttons, and submit buttons.
+
+        The standard client handler silently no-ops on ``disabled``
+        targets so the worker can't bypass UI affordances meant to be
+        user-controlled.
+
+        Args:
+            ref: Snapshot ref like ``"e42"`` from the most recent
+                ``<ui_state>``.
+        """
+        await self.send_command("click", Click(ref=ref))
+
+    async def set_input_value(
+        self,
+        ref: str,
+        value: str,
+        *,
+        replace: bool = True,
+    ) -> None:
+        """Dispatch a standard ``set_input_value`` UI command.
+
+        Convenience wrapper around ``send_command("set_input_value",
+        SetInputValue(...))``. Same pattern as the other helpers: a
+        plain instance method, not an LLM tool. Use inside custom
+        ``@tool`` bodies for filling text inputs and textareas.
+
+        With ``replace=True`` (the default) the field's existing
+        value is overwritten. With ``replace=False`` the value is
+        appended (useful for continuing a long answer in a textarea).
+
+        Args:
+            ref: Snapshot ref like ``"e42"`` of the input/textarea.
+            value: Text to write into the field.
+            replace: When True, overwrite the existing value. When
+                False, append to it. Defaults to True.
+        """
+        await self.send_command(
+            "set_input_value",
+            SetInputValue(ref=ref, value=value, replace=replace),
+        )
+
+    async def on_bus_message(self, message: BusMessage) -> None:
+        """Dispatch UI events alongside base lifecycle handling."""
+        await super().on_bus_message(message)
+
+        # Forward job lifecycle for user-facing job groups before
+        # touching anything else. This is independent of UI event
+        # handling and may fire on messages targeted at this worker
+        # (the requester) for groups it dispatched.
+        if isinstance(message, (BusJobUpdateMessage, BusJobUpdateUrgentMessage)):
+            await self._maybe_forward_job_update(message)
+            return
+        if isinstance(message, (BusJobResponseMessage, BusJobResponseUrgentMessage)):
+            await self._maybe_forward_job_completed(message)
+            return
+
+        if not isinstance(message, BusUIEventMessage):
+            return
+        if message.target and message.target != self.name:
+            return
+
+        # Reserved snapshot event: store and return without dispatch or
+        # ``<ui_event>`` injection. Apps render via ``inject_ui_state``.
+        if message.event_name == _UI_SNAPSHOT_BUS_EVENT_NAME:
+            if isinstance(message.payload, dict):
+                self._latest_snapshot = message.payload
+            return
+
+        # Reserved cancel event: route to ``cancel_job_group`` for the
+        # registered user job group. Honored only when the group was
+        # registered with ``cancellable=True``.
+        if message.event_name == _UI_CANCEL_TASK_BUS_EVENT_NAME:
+            await self._handle_cancel_job_event(message)
+            return
+
+        await self._handle_ui_event(message)
+
+    @property
+    def current_job(self) -> BusJobRequestMessage | None:
+        """The job this worker is currently processing, or ``None`` when idle.
+
+        Set when ``respond_with_llm`` starts and cleared when the job
+        completes. Lets ``@tool`` methods inspect the in-flight job
+        without threading the message through every call.
+        """
+        return self._current_job
+
+    @job(name="respond", sequential=True)
+    async def _respond_job(self, message: BusJobRequestMessage) -> None:
+        await self.respond_with_llm(message)
+
+    def render_query(self, message: BusJobRequestMessage) -> str:
+        """Extract the user's query text from a job request.
+
+        Override to read a different payload shape. The returned string
+        is appended to the LLM context as a user message before the LLM
+        runs. The default reads ``payload["query"]``.
+
+        Args:
+            message: The inbound job request.
+
+        Returns:
+            The query text to feed into the LLM.
+        """
+        return (message.payload or {}).get("query", "")
+
+    async def respond_with_llm(self, message: BusJobRequestMessage) -> None:
+        """Run one LLM turn for a job and respond when a tool completes it.
+
+        Records the in-flight job for ``respond_to_job`` to close out,
+        clears the LLM context when ``keep_history=False``, injects the
+        current snapshot when ``auto_inject_ui_state=True``, appends the
+        rendered query, and runs the LLM. Then blocks until a ``@tool``
+        calls ``respond_to_job``, and sends that result as the job
+        response.
+
+        This is the body of the built-in ``@job(name="respond",
+        sequential=True)`` handler. Because it spans the full LLM
+        round-trip, the framework's per-name lock serializes same-name
+        ``respond`` requests across the whole turn (see "Single-flight
+        job semantics" in the class docstring). Subclasses that want a
+        differently named LLM job can register their own
+        ``@job(name=..., sequential=True)`` handler that calls this.
+        """
+        self._current_job = message
+        self._pending = asyncio.get_running_loop().create_future()
+        try:
+            if not self._keep_history:
+                await self.reset_context()
+            if self._auto_inject_ui_state:
+                await self.inject_ui_state()
+            await self.queue_frame(
+                LLMMessagesAppendFrame(
+                    messages=[{"role": "user", "content": self.render_query(message)}],
+                    run_llm=True,
+                )
+            )
+            result = await self._pending
+            await self.send_job_response(
+                message.job_id, response=result["response"], status=result["status"]
+            )
+        finally:
+            self._current_job = None
+            self._pending = None
+
+    async def reset_context(self) -> None:
+        """Clear the LLM conversation history.
+
+        Replaces all messages in the running context with an empty
+        list. The system prompt (set on the LLM service via
+        ``system_instruction``) is unaffected. Messages pre-seeded
+        via ``context=`` on construction live in the same mutable
+        message list and ARE cleared, so anything that needs to
+        survive resets (durable instructions, prompt guides) belongs
+        in ``system_instruction`` rather than in seed messages.
+
+        Implemented via ``LLMMessagesUpdateFrame``, which the context
+        aggregator processes in pipeline order alongside other
+        context-modifying frames. Apps in ``keep_history=True`` mode
+        call this when they want to deliberately start over (e.g.,
+        the user said "start fresh"). Apps in the default
+        ``keep_history=False`` mode don't need to call this; the
+        per-job reset runs automatically.
+        """
+        await self.queue_frame(LLMMessagesUpdateFrame(messages=[], run_llm=False))
+
+    async def respond_to_job(
+        self,
+        response: dict | None = None,
+        *,
+        speak: str | None = None,
+        status: JobStatus = JobStatus.COMPLETED,
+    ) -> None:
+        """Complete the in-flight job this worker is processing.
+
+        Hands ``response`` (with ``speak`` merged in) to the
+        ``respond_with_llm`` handler waiting on the current job, which
+        sends it as the job response. Typically called from a ``@tool``
+        once the worker has decided how to answer.
+
+        ``speak`` is the convention the SDK demos use for "text the
+        voice worker should hand verbatim to TTS". When provided it's
+        merged into the response dict as ``{"speak": speak}``.
+        Apps that don't follow the convention can pass a fully formed
+        ``response`` dict instead and leave ``speak`` unset.
+
+        No-op when there is no job in flight (e.g. the tool was
+        invoked outside a job dispatch) or when the job has already
+        been responded to.
+
+        Args:
+            response: Result data dict. Merged with the ``speak`` key
+                when ``speak`` is provided.
+            speak: Optional short text for verbatim TTS. Omit (or pass
+                ``None``) to leave the response without a ``speak``
+                key, signaling to the voice worker that this turn
+                completes silently.
+            status: Completion status. Defaults to
+                ``JobStatus.COMPLETED``.
+        """
+        pending = self._pending
+        if pending is None or pending.done():
+            return
+        payload: dict = dict(response) if response else {}
+        if speak is not None:
+            payload["speak"] = speak
+        pending.set_result({"response": payload, "status": status})
+
+    def user_job_group(
+        self,
+        *worker_names: str,
+        name: str | None = None,
+        payload: dict | None = None,
+        timeout: float | None = None,
+        cancel_on_error: bool = True,
+        label: str | None = None,
+        cancellable: bool = True,
+    ) -> UserJobGroupContext:
+        """Dispatch a job group whose lifecycle is forwarded to the client.
+
+        Behaves exactly like ``job_group(...)`` but additionally emits
+        ``ui-task`` envelopes that the client's task reducer consumes.
+        Use this for any user-initiated work that fans out to workers
+        and that the user should be able to see (and optionally cancel)
+        on screen.
+
+        Workers don't need to change. Any ``send_job_update`` they
+        emit against the group's ``job_id`` is forwarded automatically
+        as a ``task_update`` envelope, and their final response is
+        forwarded as ``task_completed``.
+
+        Args:
+            *worker_names: Names of the workers to send the job to.
+            name: Optional job name for routing to named ``@job``
+                handlers on the workers.
+            payload: Optional structured data describing the work.
+            timeout: Optional timeout in seconds covering both the
+                ready-wait and job execution.
+            cancel_on_error: Whether to cancel the group if a worker
+                errors. Defaults to True.
+            label: Optional human-readable label surfaced to the
+                client. The client UI uses it to title the in-flight
+                task card.
+            cancellable: Whether the client may request cancellation
+                of this group via the reserved ``__cancel_task``
+                event. Defaults to True.
+
+        Returns:
+            A ``UserJobGroupContext`` to use with ``async with``.
+
+        Example::
+
+            async with self.user_job_group(
+                "researcher_a", "researcher_b",
+                payload={"query": query},
+                label=f"Research: {query}",
+            ) as tg:
+                async for event in tg:
+                    ...
+        """
+        for worker_name in worker_names:
+            if not isinstance(worker_name, str):
+                raise TypeError(
+                    f"{self} Expected worker name as str, got {type(worker_name).__name__}"
+                )
+        return UserJobGroupContext(
+            self,
+            worker_names,
+            name=name,
+            payload=payload,
+            timeout=timeout,
+            cancel_on_error=cancel_on_error,
+            label=label,
+            cancellable=cancellable,
+        )
+
+    async def start_user_job_group(
+        self,
+        *worker_names: str,
+        name: str | None = None,
+        payload: dict | None = None,
+        timeout: float | None = None,
+        cancel_on_error: bool = True,
+        label: str | None = None,
+        cancellable: bool = True,
+    ) -> str:
+        """Fire-and-forget version of ``user_job_group``.
+
+        Dispatches a user job group in the background and returns
+        immediately. The SDK manages the asyncio task that holds the
+        ``user_job_group`` context open while workers run, so callers
+        don't need to spawn one themselves. All lifecycle events
+        (``group_started``, ``task_update``, ``task_completed``,
+        ``group_completed``) are forwarded to the client as usual.
+
+        Use this when an LLM ``@tool`` body wants to kick off a job
+        group and return immediately so the voice worker unblocks.
+        Use ``user_job_group`` (the context manager) when the caller
+        wants to consume worker events inline (``async for event in
+        tg``) or react to results before returning.
+
+        Worker exceptions inside the dispatched context are logged
+        but do not propagate; cancellation works the same way it
+        does for the context-manager form (the user clicks Cancel,
+        the SDK turns it into ``cancel_job_group``).
+
+        Args:
+            *worker_names: Names of the workers to send the job to.
+            name: Optional job name for routing to named ``@job``
+                handlers on the workers.
+            payload: Optional structured data describing the work.
+            timeout: Optional timeout in seconds covering both the
+                ready-wait and job execution.
+            cancel_on_error: Whether to cancel the group if a worker
+                errors. Defaults to True.
+            label: Optional human-readable label surfaced to the
+                client. The client UI uses it to title the in-flight
+                task card.
+            cancellable: Whether the client may request cancellation
+                of this group via the reserved ``__cancel_task``
+                event. Defaults to True.
+
+        Returns:
+            The ``job_id`` of the dispatched group. Useful if the
+            caller wants to track it (e.g. to cancel programmatically
+            via ``cancel_job_group(job_id)``).
+
+        Example::
+
+            @tool
+            async def reply(self, params, answer, research_query=None):
+                if research_query:
+                    await self.start_user_job_group(
+                        "wikipedia", "news", "scholar",
+                        payload={"query": research_query},
+                        label=f"Research: {research_query}",
+                    )
+                await self.respond_to_job(speak=answer)
+                await params.result_callback(None)
+        """
+        ctx = self.user_job_group(
+            *worker_names,
+            name=name,
+            payload=payload,
+            timeout=timeout,
+            cancel_on_error=cancel_on_error,
+            label=label,
+            cancellable=cancellable,
+        )
+        # Enter the context now so the caller has a valid job_id and
+        # the ``group_started`` envelope has fired before we return.
+        # The body and exit run in a background asyncio task so the
+        # caller doesn't await worker completion.
+        await ctx.__aenter__()
+        job_id = ctx.job_id
+
+        async def _run_to_completion() -> None:
+            # Drain the event stream so __aexit__ sees a fully
+            # consumed group, matching what ``async with ... : pass``
+            # does. Cancellation and worker errors are expected exits
+            # for fire-and-forget groups: the client already learned
+            # via the group_completed envelope, so we log at debug.
+            iteration_exc: BaseException | None = None
+            try:
+                async for _ in ctx:
+                    pass
+            except Exception as e:
+                iteration_exc = e
+
+            try:
+                if iteration_exc is None:
+                    await ctx.__aexit__(None, None, None)
+                else:
+                    await ctx.__aexit__(
+                        type(iteration_exc),
+                        iteration_exc,
+                        iteration_exc.__traceback__,
+                    )
+            except JobGroupError as e:
+                logger.debug(
+                    f"UIWorker '{self.name}': background user job group {job_id} ended: {e}"
+                )
+            except Exception as e:
+                logger.warning(
+                    f"UIWorker '{self.name}': background user job group {job_id} failed: {e}"
+                )
+
+        self.create_task(
+            _run_to_completion(),
+            f"{self.name}::user_job_group::{job_id}",
+        )
+        return job_id
+
+    def _register_user_job_group(
+        self,
+        *,
+        job_id: str,
+        worker_names: list[str],
+        label: str | None,
+        cancellable: bool,
+    ) -> None:
+        """Register an in-flight user job group for lifecycle forwarding.
+
+        Called from ``UserJobGroupContext.__aenter__``. Subsequent
+        ``BusJobUpdateMessage`` / ``BusJobResponseMessage`` whose
+        ``job_id`` matches this entry will be forwarded to the client.
+        """
+        if job_id in self._user_job_groups:
+            logger.warning(
+                f"UIWorker '{self.name}': user job group {job_id} already registered; overwriting"
+            )
+        self._user_job_groups[job_id] = _UserJobGroupRegistration(
+            worker_names=list(worker_names),
+            label=label,
+            cancellable=cancellable,
+        )
+
+    def _unregister_user_job_group(self, job_id: str) -> None:
+        """Remove a user job group from the forwarding registry.
+
+        Called from ``UserJobGroupContext.__aexit__``. After this,
+        late-arriving updates or responses for the group are not
+        forwarded.
+        """
+        self._user_job_groups.pop(job_id, None)
+
+    async def _maybe_forward_job_update(
+        self, message: BusJobUpdateMessage | BusJobUpdateUrgentMessage
+    ) -> None:
+        """Forward a worker update for a registered user job group.
+
+        No-op if the message's ``job_id`` is not registered.
+        """
+        if message.job_id not in self._user_job_groups:
+            return
+        await self.send_bus_message(
+            BusUITaskUpdateMessage(
+                source=self.name,
+                target=None,
+                task_id=message.job_id,
+                agent_name=message.source,
+                data=message.update,
+                at=int(time.time() * 1000),
+            )
+        )
+
+    async def _maybe_forward_job_completed(
+        self, message: BusJobResponseMessage | BusJobResponseUrgentMessage
+    ) -> None:
+        """Forward a worker response for a registered user job group.
+
+        No-op if the message's ``job_id`` is not registered.
+        """
+        if message.job_id not in self._user_job_groups:
+            return
+        await self.send_bus_message(
+            BusUITaskCompletedMessage(
+                source=self.name,
+                target=None,
+                task_id=message.job_id,
+                agent_name=message.source,
+                status=str(message.status),
+                response=message.response,
+                at=int(time.time() * 1000),
+            )
+        )
+
+    async def _handle_cancel_job_event(self, message: BusUIEventMessage) -> None:
+        """Translate a client ``__cancel_task`` event into ``cancel_job_group``.
+
+        Looks up the registered group and calls
+        ``cancel_job_group(job_id, reason)``. Ignores the request
+        silently if the group is unknown or was registered with
+        ``cancellable=False``.
+        """
+        payload = message.payload if isinstance(message.payload, dict) else {}
+        job_id = payload.get("task_id")
+        if not isinstance(job_id, str) or not job_id:
+            logger.warning(
+                f"UIWorker '{self.name}': received {_UI_CANCEL_TASK_BUS_EVENT_NAME} "
+                "with no task_id; ignoring"
+            )
+            return
+        registration = self._user_job_groups.get(job_id)
+        if registration is None:
+            logger.debug(
+                f"UIWorker '{self.name}': {_UI_CANCEL_TASK_BUS_EVENT_NAME} for "
+                f"unknown task_id {job_id}; ignoring"
+            )
+            return
+        if not registration.cancellable:
+            logger.debug(
+                f"UIWorker '{self.name}': {_UI_CANCEL_TASK_BUS_EVENT_NAME} for "
+                f"non-cancellable group {job_id}; ignoring"
+            )
+            return
+        reason = payload.get("reason")
+        if reason is not None and not isinstance(reason, str):
+            reason = None
+        await self.cancel_job_group(job_id, reason=reason or "cancelled by user")
+
+    def render_ui_state(self) -> str:
+        """Render the latest accessibility snapshot as a ``<ui_state>`` block.
+
+        Produces Playwright-MCP-style indented text with stable element
+        refs. Apps inject the output via ``inject_ui_state()`` when they
+        want the LLM to see what's on screen.
+
+        When the snapshot carries a current text selection, a nested
+        ``<selection ref="...">...</selection>`` block is appended
+        inside ``<ui_state>`` so the LLM can resolve deictic references
+        ("this paragraph", "what I selected") against on-page content.
+
+        Override to customize the rendered form. Returns an empty
+        string if no snapshot has been received yet.
+        """
+        if not self._latest_snapshot:
+            return ""
+        root = self._latest_snapshot.get("root")
+        if not isinstance(root, dict):
+            return ""
+        lines = ["<ui_state>"]
+        _render_node(root, depth=0, lines=lines)
+        selection = self._latest_snapshot.get("selection")
+        if isinstance(selection, dict):
+            _render_selection(selection, lines)
+        lines.append("</ui_state>")
+        return "\n".join(lines)
+
+    def visible_nodes(self) -> list[dict[str, Any]]:
+        """Return the snapshot nodes the user is currently looking at.
+
+        Walks the latest snapshot and returns a flat list of every
+        node whose ``state`` does not contain ``"offscreen"``. Useful
+        for apps that want to reason about visible interactables
+        without scanning the rendered ``<ui_state>`` text.
+
+        Ordering follows a depth-first traversal of the snapshot,
+        matching the order in ``<ui_state>``. Returns an empty list
+        when no snapshot has been received yet.
+        """
+        if not self._latest_snapshot:
+            return []
+        root = self._latest_snapshot.get("root")
+        if not isinstance(root, dict):
+            return []
+        out: list[dict[str, Any]] = []
+        _collect_visible(root, out)
+        return out
+
+    async def inject_ui_state(self) -> None:
+        """Append the latest ``<ui_state>`` block to the LLM context.
+
+        No-op when no snapshot has been received. Frame has
+        ``run_llm=False`` — the snapshot is context, not a user turn.
+        """
+        content = self.render_ui_state()
+        if not content:
+            return
+        await self.queue_frame(
+            LLMMessagesAppendFrame(
+                messages=[{"role": "developer", "content": content}],
+                run_llm=False,
+            )
+        )
+
+    def render_ui_event(self, message: BusUIEventMessage) -> str:
+        """Render a UI event as a string for LLM context injection.
+
+        Override to customize the injected content. The default wraps
+        the event in a single ``<ui_event>`` XML tag with a ``name``
+        attribute and a JSON-encoded payload as inner text.
+
+        Args:
+            message: The UI event to render.
+
+        Returns:
+            A string to append to the LLM context as a developer message.
+        """
+        payload_repr = json.dumps(message.payload, default=str)
+        return f'<ui_event name="{message.event_name}">{payload_repr}</ui_event>'
+
+    async def _handle_ui_event(self, message: BusUIEventMessage) -> None:
+        """Inject the event into LLM context, then dispatch to the handler.
+
+        Injection runs synchronously first so the ``<ui_event>``
+        developer message lands in the context before any side effects
+        the handler triggers. The matching ``@on_ui_event`` handler
+        then runs in its own asyncio task so the bus dispatcher isn't
+        held open while the handler awaits downstream work (job
+        requests, network calls). Events with no registered handler
+        are a no-op after injection.
+        """
+        if self._inject_events:
+            content = self.render_ui_event(message)
+            if content:
+                await self.queue_frame(
+                    LLMMessagesAppendFrame(
+                        messages=[{"role": "developer", "content": content}],
+                        run_llm=False,
+                    )
+                )
+
+        handler = self._ui_event_handlers.get(message.event_name)
+        if handler is None:
+            return
+
+        # Handlers run in their own asyncio task so the bus dispatcher
+        # is never held open while a handler awaits downstream work
+        # (job requests, network calls, etc.). Same pattern as ``@job``.
+        self.create_task(
+            handler(message),
+            f"{self.name}::ui_event_{message.event_name}",
+        )
+
+
+def _collect_visible(node: dict[str, Any], out: list[dict[str, Any]]) -> None:
+    """Depth-first collect nodes whose state does not include offscreen."""
+    state = node.get("state")
+    is_offscreen = isinstance(state, list) and "offscreen" in state
+    if not is_offscreen:
+        out.append(node)
+    children = node.get("children")
+    if isinstance(children, list):
+        for child in children:
+            if isinstance(child, dict):
+                _collect_visible(child, out)
+
+
+def _render_node(node: dict[str, Any], *, depth: int, lines: list[str]) -> None:
+    """Render one A11yNode dict as Playwright-MCP-style indented text.
+
+    Format per node::
+
+        - role "name" [level=N] [cols=N] [rows=N] [state1] [state2] [ref=eN]:
+
+    Trailing ``:`` when the node has children. ``name``, ``level``,
+    grid dims, and state tags are emitted only when present on the
+    node.
+    """
+    role = node.get("role", "generic")
+    name = node.get("name")
+    value = node.get("value")
+    state = node.get("state") or []
+    level = node.get("level")
+    colcount = node.get("colcount")
+    rowcount = node.get("rowcount")
+    ref = node.get("ref", "")
+    children = node.get("children") or []
+
+    parts: list[str] = [f"- {role}"]
+    if isinstance(name, str) and name:
+        parts.append(f'"{name}"')
+    if isinstance(value, str) and value:
+        parts.append(f'= "{value}"')
+    if isinstance(level, int):
+        parts.append(f"[level={level}]")
+    if isinstance(colcount, int):
+        parts.append(f"[cols={colcount}]")
+    if isinstance(rowcount, int):
+        parts.append(f"[rows={rowcount}]")
+    if isinstance(state, list):
+        for s in state:
+            if isinstance(s, str) and s:
+                parts.append(f"[{s}]")
+    if isinstance(ref, str) and ref:
+        parts.append(f"[ref={ref}]")
+
+    indent = "  " * depth
+    line = indent + " ".join(parts)
+    if children:
+        line += ":"
+    lines.append(line)
+
+    if isinstance(children, list):
+        for child in children:
+            if isinstance(child, dict):
+                _render_node(child, depth=depth + 1, lines=lines)
+
+
+def _render_selection(selection: dict[str, Any], lines: list[str]) -> None:
+    """Render an ``A11ySelection`` dict as a ``<selection>`` block.
+
+    Emitted at the root of ``<ui_state>`` (no leading indent) so the
+    LLM can spot it without parsing the tree::
+
+        <selection ref="e42">
+        the actual selected text
+        </selection>
+
+    No-op when the selection lacks a ``ref`` or ``text``.
+    """
+    ref = selection.get("ref")
+    text = selection.get("text")
+    if not isinstance(ref, str) or not ref:
+        return
+    if not isinstance(text, str) or not text:
+        return
+    lines.append(f'<selection ref="{ref}">')
+    lines.append(text)
+    lines.append("</selection>")
diff --git a/tests/test_ui_commands.py b/tests/test_ui_commands.py
new file mode 100644
index 000000000..2db9052a8
--- /dev/null
+++ b/tests/test_ui_commands.py
@@ -0,0 +1,231 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for UIWorker.send_command and standard command payload models."""
+
+import unittest
+from unittest.mock import MagicMock
+
+from pipecat.bus.ui_messages import BusUICommandMessage
+from pipecat.processors.frameworks.rtvi.models import (
+    Click,
+    Focus,
+    Highlight,
+    Navigate,
+    ScrollTo,
+    SelectText,
+    SetInputValue,
+    Toast,
+)
+from pipecat.workers.ui import UIWorker
+
+
+def _make_worker():
+    """A UIWorker whose ``send_bus_message`` captures sent commands.
+
+    ``send_command`` publishes via ``send_bus_message``; replacing it
+    avoids needing an attached bus for these unit tests.
+    """
+    worker = UIWorker("ui", llm=MagicMock(), active=False)
+    sent: list[BusUICommandMessage] = []
+
+    async def _record(message):
+        if isinstance(message, BusUICommandMessage):
+            sent.append(message)
+
+    worker.send_bus_message = _record  # type: ignore[method-assign]
+    return worker, sent
+
+
+class TestSendCommand(unittest.IsolatedAsyncioTestCase):
+    async def test_serializes_pydantic_payload_via_model_dump(self):
+        worker, sent = _make_worker()
+
+        await worker.send_command("toast", Toast(title="Saved", subtitle="Favorites"))
+
+        self.assertEqual(len(sent), 1)
+        cmd = sent[0]
+        self.assertEqual(cmd.source, "ui")
+        self.assertIsNone(cmd.target)
+        self.assertEqual(cmd.command_name, "toast")
+        self.assertEqual(
+            cmd.payload,
+            {
+                "title": "Saved",
+                "subtitle": "Favorites",
+                "description": None,
+                "image_url": None,
+                "duration_ms": None,
+            },
+        )
+
+    async def test_forwards_dict_payload_as_is(self):
+        worker, sent = _make_worker()
+
+        await worker.send_command("app_specific", {"foo": 1, "bar": [1, 2, 3]})
+
+        self.assertEqual(sent[0].command_name, "app_specific")
+        self.assertEqual(sent[0].payload, {"foo": 1, "bar": [1, 2, 3]})
+
+    async def test_none_payload_becomes_empty_dict(self):
+        worker, sent = _make_worker()
+
+        await worker.send_command("ping")
+
+        self.assertEqual(sent[0].payload, {})
+
+    async def test_dict_payload_for_apps_with_custom_command_names(self):
+        worker, sent = _make_worker()
+
+        await worker.send_command("navigate", {"view": "home"})
+        self.assertEqual(sent[0].payload, {"view": "home"})
+
+
+class TestStandardCommands(unittest.IsolatedAsyncioTestCase):
+    async def test_toast_payload_shape(self):
+        worker, sent = _make_worker()
+        await worker.send_command(
+            "toast",
+            Toast(
+                title="Now playing",
+                subtitle="Nirvana",
+                description="Smells Like Teen Spirit",
+                image_url="https://example.com/cover.jpg",
+                duration_ms=3000,
+            ),
+        )
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "title": "Now playing",
+                "subtitle": "Nirvana",
+                "description": "Smells Like Teen Spirit",
+                "image_url": "https://example.com/cover.jpg",
+                "duration_ms": 3000,
+            },
+        )
+
+    async def test_navigate_payload_shape(self):
+        worker, sent = _make_worker()
+        await worker.send_command("navigate", Navigate(view="detail", params={"id": "42"}))
+        self.assertEqual(sent[0].payload, {"view": "detail", "params": {"id": "42"}})
+
+    async def test_scroll_to_payload_shape_by_target_id(self):
+        worker, sent = _make_worker()
+        await worker.send_command(
+            "scroll_to", ScrollTo(target_id="new_releases", behavior="smooth")
+        )
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": None, "target_id": "new_releases", "behavior": "smooth"},
+        )
+
+    async def test_scroll_to_payload_shape_by_ref(self):
+        worker, sent = _make_worker()
+        await worker.send_command("scroll_to", ScrollTo(ref="e42", behavior="smooth"))
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": "e42", "target_id": None, "behavior": "smooth"},
+        )
+
+    async def test_highlight_payload_shape(self):
+        worker, sent = _make_worker()
+        await worker.send_command("highlight", Highlight(target_id="play_btn", duration_ms=1000))
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": None, "target_id": "play_btn", "duration_ms": 1000},
+        )
+
+    async def test_focus_payload_shape(self):
+        worker, sent = _make_worker()
+        await worker.send_command("focus", Focus(target_id="search_input"))
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": None, "target_id": "search_input"},
+        )
+
+    async def test_focus_payload_shape_by_ref(self):
+        worker, sent = _make_worker()
+        await worker.send_command("focus", Focus(ref="e7"))
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": "e7", "target_id": None},
+        )
+
+    async def test_select_text_payload_whole_element(self):
+        worker, sent = _make_worker()
+        await worker.send_command("select_text", SelectText(ref="e42"))
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e42",
+                "target_id": None,
+                "start_offset": None,
+                "end_offset": None,
+            },
+        )
+
+    async def test_select_text_payload_with_offsets(self):
+        worker, sent = _make_worker()
+        await worker.send_command(
+            "select_text",
+            SelectText(ref="e42", start_offset=5, end_offset=12),
+        )
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e42",
+                "target_id": None,
+                "start_offset": 5,
+                "end_offset": 12,
+            },
+        )
+
+    async def test_set_input_value_payload_replace_default(self):
+        worker, sent = _make_worker()
+        await worker.send_command(
+            "set_input_value",
+            SetInputValue(ref="e7", value="hello"),
+        )
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e7",
+                "target_id": None,
+                "value": "hello",
+                "replace": True,
+            },
+        )
+
+    async def test_set_input_value_payload_append(self):
+        worker, sent = _make_worker()
+        await worker.send_command(
+            "set_input_value",
+            SetInputValue(ref="e7", value=" world", replace=False),
+        )
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e7",
+                "target_id": None,
+                "value": " world",
+                "replace": False,
+            },
+        )
+
+    async def test_click_payload_by_ref(self):
+        worker, sent = _make_worker()
+        await worker.send_command("click", Click(ref="e42"))
+        self.assertEqual(sent[0].payload, {"ref": "e42", "target_id": None})
+
+    async def test_click_payload_by_target_id(self):
+        worker, sent = _make_worker()
+        await worker.send_command("click", Click(target_id="submit"))
+        self.assertEqual(sent[0].payload, {"ref": None, "target_id": "submit"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_ui_job_lifecycle.py b/tests/test_ui_job_lifecycle.py
new file mode 100644
index 000000000..2d7c3f921
--- /dev/null
+++ b/tests/test_ui_job_lifecycle.py
@@ -0,0 +1,366 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for the UIWorker user-job-group lifecycle.
+
+Covers:
+- ``UIWorker.on_bus_message`` forwarding of worker job updates/responses
+  for registered user job groups as ``BusUITask*`` carriers.
+- The reserved ``__cancel_task`` client event routing to
+  ``cancel_job_group``.
+- ``UserJobGroupContext`` publishing ``group_started`` / ``group_completed``
+  envelopes and (de)registering the group on the worker.
+"""
+
+import asyncio
+import unittest
+from unittest.mock import AsyncMock, MagicMock
+
+from pipecat.bus.messages import BusJobResponseMessage, BusJobUpdateMessage
+from pipecat.bus.ui_messages import (
+    _UI_CANCEL_TASK_BUS_EVENT_NAME,
+    BusUIEventMessage,
+    BusUITaskCompletedMessage,
+    BusUITaskGroupCompletedMessage,
+    BusUITaskGroupStartedMessage,
+    BusUITaskUpdateMessage,
+)
+from pipecat.frames.frames import LLMMessagesAppendFrame
+from pipecat.pipeline.job_context import JobGroup, JobStatus
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
+from pipecat.workers.ui import UIWorker
+
+
+async def _make_solo_worker(**kwargs) -> UIWorker:
+    """A UIWorker with a task manager and a ``queue_frame`` spy.
+
+    Suitable for testing forwarding logic by directly invoking
+    ``on_bus_message`` and asserting on captured ``send_bus_message``
+    calls.
+    """
+    worker = UIWorker("ui", llm=MagicMock(), active=False, **kwargs)
+    tm = TaskManager()
+    tm.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
+    worker._task_manager = tm
+
+    recorded: list = []
+
+    async def _record(frame, direction=FrameDirection.DOWNSTREAM):
+        recorded.append(frame)
+
+    worker.queue_frame = _record  # type: ignore[method-assign]
+    worker._recorded = recorded  # type: ignore[attr-defined]
+    return worker
+
+
+class TestUIWorkerForwarding(unittest.IsolatedAsyncioTestCase):
+    async def test_unregistered_job_update_is_not_forwarded(self):
+        worker = await _make_solo_worker()
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusJobUpdateMessage(
+                source="worker", target=worker.name, job_id="t-unknown", update={"x": 1}
+            )
+        )
+
+        forwarded = [
+            c.args[0]
+            for c in worker.send_bus_message.await_args_list
+            if isinstance(c.args[0], BusUITaskUpdateMessage)
+        ]
+        self.assertEqual(forwarded, [])
+
+    async def test_registered_job_update_is_forwarded(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["worker"], label="hello", cancellable=True
+        )
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusJobUpdateMessage(
+                source="worker",
+                target=worker.name,
+                job_id="t1",
+                update={"kind": "tool_call", "tool": "WebSearch"},
+            )
+        )
+
+        forwarded = [
+            c.args[0]
+            for c in worker.send_bus_message.await_args_list
+            if isinstance(c.args[0], BusUITaskUpdateMessage)
+        ]
+        self.assertEqual(len(forwarded), 1)
+        self.assertEqual(forwarded[0].task_id, "t1")
+        self.assertEqual(forwarded[0].agent_name, "worker")
+        self.assertEqual(forwarded[0].data, {"kind": "tool_call", "tool": "WebSearch"})
+
+    async def test_registered_job_response_is_forwarded(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["worker"], label=None, cancellable=True
+        )
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusJobResponseMessage(
+                source="worker",
+                target=worker.name,
+                job_id="t1",
+                status=JobStatus.COMPLETED,
+                response={"answer": 42},
+            )
+        )
+
+        forwarded = [
+            c.args[0]
+            for c in worker.send_bus_message.await_args_list
+            if isinstance(c.args[0], BusUITaskCompletedMessage)
+        ]
+        self.assertEqual(len(forwarded), 1)
+        self.assertEqual(forwarded[0].task_id, "t1")
+        self.assertEqual(forwarded[0].agent_name, "worker")
+        self.assertEqual(forwarded[0].status, "completed")
+        self.assertEqual(forwarded[0].response, {"answer": 42})
+
+    async def test_response_status_serializes_for_cancelled_and_error(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["w"], label=None, cancellable=True
+        )
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusJobResponseMessage(
+                source="w", target=worker.name, job_id="t1", status=JobStatus.CANCELLED
+            )
+        )
+        await worker.on_bus_message(
+            BusJobResponseMessage(
+                source="w", target=worker.name, job_id="t1", status=JobStatus.ERROR
+            )
+        )
+
+        statuses = [
+            c.args[0].status
+            for c in worker.send_bus_message.await_args_list
+            if isinstance(c.args[0], BusUITaskCompletedMessage)
+        ]
+        self.assertEqual(statuses, ["cancelled", "error"])
+
+
+class TestCancelJobEvent(unittest.IsolatedAsyncioTestCase):
+    async def test_cancel_event_routes_to_cancel_job_group(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["w"], label=None, cancellable=True
+        )
+        worker.cancel_job_group = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload={"task_id": "t1", "reason": "user clicked cancel"},
+            )
+        )
+
+        worker.cancel_job_group.assert_awaited_once_with("t1", reason="user clicked cancel")
+
+    async def test_cancel_event_default_reason_when_omitted(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["w"], label=None, cancellable=True
+        )
+        worker.cancel_job_group = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload={"task_id": "t1"},
+            )
+        )
+
+        worker.cancel_job_group.assert_awaited_once()
+        self.assertEqual(worker.cancel_job_group.await_args.kwargs["reason"], "cancelled by user")
+
+    async def test_non_cancellable_group_is_ignored(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["w"], label=None, cancellable=False
+        )
+        worker.cancel_job_group = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload={"task_id": "t1"},
+            )
+        )
+
+        worker.cancel_job_group.assert_not_awaited()
+
+    async def test_unknown_job_id_is_ignored(self):
+        worker = await _make_solo_worker()
+        worker.cancel_job_group = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload={"task_id": "nope"},
+            )
+        )
+
+        worker.cancel_job_group.assert_not_awaited()
+
+    async def test_missing_or_bad_payload_is_ignored(self):
+        worker = await _make_solo_worker()
+        worker.cancel_job_group = AsyncMock()  # type: ignore[method-assign]
+
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload=None,
+            )
+        )
+        await worker.on_bus_message(
+            BusUIEventMessage(
+                source="bridge",
+                target=worker.name,
+                event_name=_UI_CANCEL_TASK_BUS_EVENT_NAME,
+                payload={"task_id": 42},
+            )
+        )
+
+        worker.cancel_job_group.assert_not_awaited()
+
+
+class TestForwardingDoesNotInjectLLMContext(unittest.IsolatedAsyncioTestCase):
+    async def test_job_update_forwarding_does_not_queue_append_frames(self):
+        worker = await _make_solo_worker()
+        worker._register_user_job_group(
+            job_id="t1", worker_names=["w"], label=None, cancellable=True
+        )
+
+        await worker.on_bus_message(
+            BusJobUpdateMessage(source="w", target=worker.name, job_id="t1", update={"x": 1})
+        )
+
+        appends = [f for f in worker._recorded if isinstance(f, LLMMessagesAppendFrame)]
+        self.assertEqual(appends, [])
+
+
+def _stub_job_group(worker, job_id="t1", worker_names=("w1",)):
+    """Make ``create_job_group_and_request_job`` return a self-completing group.
+
+    The group completes on the next loop tick so both the context-manager
+    and fire-and-forget paths terminate without a running bus or workers.
+    """
+
+    async def _fake_create(names, *, name=None, payload=None, timeout=None, cancel_on_error=True):
+        group = JobGroup(job_id=job_id, worker_names=set(names))
+
+        async def _finish():
+            # Yield so JobGroupContext.__aenter__ can set event_queue first.
+            await asyncio.sleep(0)
+            group.complete()
+
+        asyncio.create_task(_finish())
+        return group
+
+    worker.create_job_group_and_request_job = _fake_create  # type: ignore[method-assign]
+
+
+class TestUserJobGroupContext(unittest.IsolatedAsyncioTestCase):
+    async def test_context_publishes_started_and_completed(self):
+        worker = await _make_solo_worker()
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+        _stub_job_group(worker)
+
+        async with worker.user_job_group("w1", label="My research") as tg:
+            self.assertEqual(tg.job_id, "t1")
+            self.assertIn("t1", worker._user_job_groups)
+
+        self.assertNotIn("t1", worker._user_job_groups)
+
+        kinds = [type(c.args[0]).__name__ for c in worker.send_bus_message.await_args_list]
+        self.assertEqual(
+            kinds,
+            ["BusUITaskGroupStartedMessage", "BusUITaskGroupCompletedMessage"],
+        )
+
+        started = worker.send_bus_message.await_args_list[0].args[0]
+        self.assertIsInstance(started, BusUITaskGroupStartedMessage)
+        self.assertEqual(started.task_id, "t1")
+        self.assertEqual(started.agents, ["w1"])
+        self.assertEqual(started.label, "My research")
+        self.assertTrue(started.cancellable)
+
+        completed = worker.send_bus_message.await_args_list[1].args[0]
+        self.assertIsInstance(completed, BusUITaskGroupCompletedMessage)
+        self.assertEqual(completed.task_id, "t1")
+
+    async def test_non_cancellable_group_sets_flag_in_started_message(self):
+        worker = await _make_solo_worker()
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+        _stub_job_group(worker)
+
+        async with worker.user_job_group("w1", cancellable=False):
+            pass
+
+        started = worker.send_bus_message.await_args_list[0].args[0]
+        self.assertFalse(started.cancellable)
+
+    async def test_unregisters_on_exit(self):
+        worker = await _make_solo_worker()
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+        _stub_job_group(worker)
+
+        async with worker.user_job_group("w1") as tg:
+            pass
+
+        self.assertNotIn(tg.job_id, worker._user_job_groups)
+
+    async def test_start_user_job_group_returns_id_and_publishes(self):
+        worker = await _make_solo_worker()
+        worker.send_bus_message = AsyncMock()  # type: ignore[method-assign]
+        _stub_job_group(worker)
+
+        job_id = await worker.start_user_job_group("w1", label="Background work")
+        self.assertEqual(job_id, "t1")
+
+        started = worker.send_bus_message.await_args_list[0].args[0]
+        self.assertIsInstance(started, BusUITaskGroupStartedMessage)
+        self.assertEqual(started.label, "Background work")
+
+        # The background runner drains the group and publishes completion.
+        for _ in range(50):
+            await asyncio.sleep(0)
+            if any(
+                isinstance(c.args[0], BusUITaskGroupCompletedMessage)
+                for c in worker.send_bus_message.await_args_list
+            ):
+                break
+        else:
+            self.fail("group_completed envelope was not published")
+
+        self.assertNotIn("t1", worker._user_job_groups)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_ui_tools.py b/tests/test_ui_tools.py
new file mode 100644
index 000000000..64d3a807c
--- /dev/null
+++ b/tests/test_ui_tools.py
@@ -0,0 +1,418 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for ``ReplyToolMixin`` and the action helper methods on ``UIWorker``.
+
+The mixin exposes a single bundled ``reply(answer, scroll_to,
+highlight, ...)`` LLM tool whose ``answer`` argument is required. The
+helper methods (``scroll_to``, ``highlight``, ...) are plain instance
+methods on ``UIWorker`` that wrap ``send_command`` with the standard
+payload models; apps call them inside custom ``@tool`` bodies when the
+canonical ``reply`` shape doesn't fit.
+"""
+
+import unittest
+from unittest.mock import AsyncMock, MagicMock
+
+from pipecat.bus.ui_messages import BusUICommandMessage
+from pipecat.workers.llm.tool_decorator import _collect_tools
+from pipecat.workers.ui import ReplyToolMixin, UIWorker
+
+
+class _WorkerWithReply(ReplyToolMixin, UIWorker):
+    pass
+
+
+class _PlainWorker(UIWorker):
+    pass
+
+
+def _new(cls: type) -> UIWorker:
+    return cls("ui", llm=MagicMock(), active=False)
+
+
+def _capture(worker: UIWorker) -> list[BusUICommandMessage]:
+    sent: list[BusUICommandMessage] = []
+
+    async def _record(message):
+        sent.append(message)
+
+    worker.send_bus_message = _record  # type: ignore[method-assign]
+    return sent
+
+
+class TestUIWorkerActionHelpers(unittest.IsolatedAsyncioTestCase):
+    """The helper methods are plain methods, not LLM tools."""
+
+    async def test_scroll_to_helper_dispatches_command(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.scroll_to("e42")
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].command_name, "scroll_to")
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": "e42", "target_id": None, "behavior": None},
+        )
+
+    async def test_highlight_helper_dispatches_command(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.highlight("e7")
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].command_name, "highlight")
+        self.assertEqual(
+            sent[0].payload,
+            {"ref": "e7", "target_id": None, "duration_ms": None},
+        )
+
+    async def test_select_text_helper_whole_element(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.select_text("e42")
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].command_name, "select_text")
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e42",
+                "target_id": None,
+                "start_offset": None,
+                "end_offset": None,
+            },
+        )
+
+    async def test_select_text_helper_with_offsets(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.select_text("e42", start_offset=10, end_offset=25)
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e42",
+                "target_id": None,
+                "start_offset": 10,
+                "end_offset": 25,
+            },
+        )
+
+    async def test_click_helper_dispatches_command(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.click("e42")
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].command_name, "click")
+        self.assertEqual(sent[0].payload, {"ref": "e42", "target_id": None})
+
+    async def test_set_input_value_helper_default_replace(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.set_input_value("e42", "hello world")
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].command_name, "set_input_value")
+        self.assertEqual(
+            sent[0].payload,
+            {
+                "ref": "e42",
+                "target_id": None,
+                "value": "hello world",
+                "replace": True,
+            },
+        )
+
+    async def test_set_input_value_helper_append_mode(self):
+        worker = _new(_PlainWorker)
+        sent = _capture(worker)
+
+        await worker.set_input_value("e42", "more text", replace=False)
+
+        self.assertEqual(sent[0].payload["replace"], False)
+
+    async def test_helpers_are_not_llm_tools(self):
+        worker = _new(_PlainWorker)
+        tool_names = [t.__name__ for t in _collect_tools(worker)]
+        for name in ("scroll_to", "highlight", "select_text", "click", "set_input_value"):
+            self.assertNotIn(name, tool_names)
+
+
+class TestReplyToolMixin(unittest.IsolatedAsyncioTestCase):
+    async def test_mixin_exposes_reply_tool(self):
+        worker = _new(_WorkerWithReply)
+        tool_names = [t.__name__ for t in _collect_tools(worker)]
+        self.assertEqual(tool_names, ["reply"])
+
+    async def test_plain_uiworker_has_no_reply_tool(self):
+        worker = _new(_PlainWorker)
+        tool_names = [t.__name__ for t in _collect_tools(worker)]
+        self.assertNotIn("reply", tool_names)
+
+    async def test_reply_with_answer_only_terminates(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(params, answer="The Pixel 9 is from Google.")
+
+        self.assertEqual(sent, [])
+        worker.respond_to_job.assert_awaited_once_with(speak="The Pixel 9 is from Google.")
+        params.result_callback.assert_awaited_once_with(None)
+
+    async def test_reply_with_highlight_only(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="This one, the Nothing Phone 3.",
+            highlight=["e29"],
+        )
+
+        self.assertEqual([m.command_name for m in sent], ["highlight"])
+        self.assertEqual(sent[0].payload["ref"], "e29")
+        worker.respond_to_job.assert_awaited_once_with(speak="This one, the Nothing Phone 3.")
+
+    async def test_reply_with_multiple_highlights(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="Here are the Apple phones.",
+            highlight=["e5", "e8", "e47"],
+        )
+
+        self.assertEqual([m.command_name for m in sent], ["highlight"] * 3)
+        self.assertEqual([m.payload["ref"] for m in sent], ["e5", "e8", "e47"])
+        worker.respond_to_job.assert_awaited_once_with(speak="Here are the Apple phones.")
+
+    async def test_reply_with_scroll_and_highlight_runs_in_order(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="Here's the iPhone 17.",
+            scroll_to="e5",
+            highlight=["e5"],
+        )
+
+        self.assertEqual([m.command_name for m in sent], ["scroll_to", "highlight"])
+        worker.respond_to_job.assert_awaited_once_with(speak="Here's the iPhone 17.")
+
+    async def test_reply_with_select_text_only(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="Here, in this paragraph.",
+            select_text="e11",
+        )
+
+        self.assertEqual([m.command_name for m in sent], ["select_text"])
+        self.assertEqual(sent[0].payload["ref"], "e11")
+        worker.respond_to_job.assert_awaited_once_with(speak="Here, in this paragraph.")
+
+    async def test_reply_with_scroll_and_select_text(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="Here, in this paragraph.",
+            scroll_to="e11",
+            select_text="e11",
+        )
+
+        self.assertEqual([m.command_name for m in sent], ["scroll_to", "select_text"])
+
+    async def test_reply_with_fills_writes_each_input(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="Got it.",
+            fills=[
+                {"ref": "e5", "value": "Mark"},
+                {"ref": "e7", "value": "Backman"},
+            ],
+        )
+
+        self.assertEqual(
+            [m.command_name for m in sent],
+            ["set_input_value", "set_input_value"],
+        )
+        self.assertEqual(sent[0].payload["ref"], "e5")
+        self.assertEqual(sent[0].payload["value"], "Mark")
+        self.assertEqual(sent[1].payload["ref"], "e7")
+        self.assertEqual(sent[1].payload["value"], "Backman")
+
+    async def test_reply_with_click_clicks_each_in_order(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(params, answer="Submitted.", click=["e22", "e26"])
+
+        self.assertEqual([m.command_name for m in sent], ["click", "click"])
+        self.assertEqual([m.payload["ref"] for m in sent], ["e22", "e26"])
+
+    async def test_reply_with_fills_skips_invalid_entries(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="x",
+            fills=[
+                {"ref": "e5", "value": "Mark"},
+                {"ref": None, "value": "missing ref"},
+                {"value": "no ref"},
+                {"ref": "e7"},
+            ],
+        )
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].payload["ref"], "e5")
+
+    async def test_reply_with_non_dict_fill_entries_does_not_crash(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="x",
+            fills=[None, "e5", 42, {"ref": "e9", "value": "ok"}],  # type: ignore[list-item]
+        )
+
+        self.assertEqual(len(sent), 1)
+        self.assertEqual(sent[0].payload["ref"], "e9")
+        worker.respond_to_job.assert_awaited_once_with(speak="x")
+        params.result_callback.assert_awaited_once_with(None)
+
+    async def test_reply_with_non_string_highlight_refs_skipped(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="x",
+            highlight=[None, "e1", 42, "e2"],  # type: ignore[list-item]
+        )
+
+        self.assertEqual([m.payload["ref"] for m in sent], ["e1", "e2"])
+        worker.respond_to_job.assert_awaited_once_with(speak="x")
+
+    async def test_reply_with_non_string_click_refs_skipped(self):
+        worker = _new(_WorkerWithReply)
+        sent = _capture(worker)
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="x",
+            click=[None, "e1", {"ref": "e2"}, "e3"],  # type: ignore[list-item]
+        )
+
+        self.assertEqual([m.payload["ref"] for m in sent], ["e1", "e3"])
+        worker.respond_to_job.assert_awaited_once_with(speak="x")
+
+    async def test_reply_dispatches_via_helper_methods(self):
+        worker = _new(_WorkerWithReply)
+        worker.scroll_to = AsyncMock()  # type: ignore[method-assign]
+        worker.highlight = AsyncMock()  # type: ignore[method-assign]
+        worker.select_text = AsyncMock()  # type: ignore[method-assign]
+        worker.set_input_value = AsyncMock()  # type: ignore[method-assign]
+        worker.click = AsyncMock()  # type: ignore[method-assign]
+        worker.respond_to_job = AsyncMock()  # type: ignore[method-assign]
+
+        params = MagicMock()
+        params.result_callback = AsyncMock()
+
+        await worker.reply(
+            params,
+            answer="x",
+            scroll_to="e1",
+            highlight=["e2", "e3"],
+            select_text="e4",
+            fills=[{"ref": "e5", "value": "v"}],
+            click=["e6", "e7"],
+        )
+
+        worker.scroll_to.assert_awaited_once_with("e1")
+        self.assertEqual(
+            worker.highlight.await_args_list,
+            [unittest.mock.call("e2"), unittest.mock.call("e3")],
+        )
+        worker.select_text.assert_awaited_once_with("e4")
+        worker.set_input_value.assert_awaited_once_with("e5", "v")
+        self.assertEqual(
+            worker.click.await_args_list,
+            [unittest.mock.call("e6"), unittest.mock.call("e7")],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_ui_worker.py b/tests/test_ui_worker.py
new file mode 100644
index 000000000..eb9e99c1d
--- /dev/null
+++ b/tests/test_ui_worker.py
@@ -0,0 +1,805 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Tests for UIWorker dispatch, LLM-context injection, and single-flight respond."""
+
+import asyncio
+import json
+import unittest
+from unittest.mock import AsyncMock, MagicMock
+
+from pipecat.bus.messages import BusJobCancelMessage, BusJobRequestMessage
+from pipecat.bus.ui_messages import _UI_SNAPSHOT_BUS_EVENT_NAME, BusUIEventMessage
+from pipecat.frames.frames import LLMMessagesAppendFrame, LLMMessagesUpdateFrame
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
+from pipecat.workers.ui import UI_STATE_PROMPT_GUIDE, UIWorker, on_ui_event
+
+
+class _StubUIWorker(UIWorker):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.captured: list[BusUIEventMessage] = []
+
+    @on_ui_event("nav_click")
+    async def _on_nav(self, message: BusUIEventMessage) -> None:
+        self.captured.append(message)
+
+
+class _PlainWorker(UIWorker):
+    pass
+
+
+async def _make_worker(cls=_StubUIWorker, **kwargs) -> UIWorker:
+    """A UIWorker wired with a task manager and a ``queue_frame`` spy.
+
+    ``queue_frame`` is replaced with a recorder so tests can assert the
+    frames the UI logic produces without running the pipeline. The
+    respond handler sends its own response, so ``send_job_response`` is
+    mocked to let ``respond_with_llm`` run without an active job entry.
+    """
+    worker = cls("ui", llm=MagicMock(), active=False, **kwargs)
+    tm = TaskManager()
+    tm.setup(TaskManagerParams(loop=asyncio.get_running_loop()))
+    worker._task_manager = tm
+
+    recorded: list = []
+
+    async def _record(frame, direction=FrameDirection.DOWNSTREAM):
+        recorded.append(frame)
+
+    worker.queue_frame = _record  # type: ignore[method-assign]
+    worker._recorded = recorded  # type: ignore[attr-defined]
+    worker.send_job_response = AsyncMock()  # type: ignore[method-assign]
+    return worker
+
+
+async def _settle() -> None:
+    """Yield enough times for spawned task handlers to run/park."""
+    for _ in range(10):
+        await asyncio.sleep(0)
+
+
+async def _start(worker: UIWorker, message: BusJobRequestMessage) -> asyncio.Task:
+    """Start a respond turn; it parks at ``await self._pending`` until resolved."""
+    t = asyncio.create_task(worker.respond_with_llm(message))
+    await _settle()
+    return t
+
+
+def _respond_msg(job_id: str, query: str = "hi") -> BusJobRequestMessage:
+    return BusJobRequestMessage(
+        source="voice", target="ui", job_name="respond", job_id=job_id, payload={"query": query}
+    )
+
+
+async def _dispatch(worker: UIWorker, message: BusUIEventMessage) -> None:
+    await worker.on_bus_message(message)
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+
+def _append_frames(worker) -> list[LLMMessagesAppendFrame]:
+    return [f for f in worker._recorded if isinstance(f, LLMMessagesAppendFrame)]
+
+
+def _update_frames(worker) -> list[LLMMessagesUpdateFrame]:
+    return [f for f in worker._recorded if isinstance(f, LLMMessagesUpdateFrame)]
+
+
+class TestUIWorkerDispatch(unittest.IsolatedAsyncioTestCase):
+    async def test_dispatches_to_matching_on_ui_event_handler(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        self.assertEqual(len(worker.captured), 1)
+        self.assertEqual(worker.captured[0].event_name, "nav_click")
+        self.assertEqual(worker.captured[0].payload, {"view": "home"})
+
+    async def test_unknown_event_name_does_not_raise(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="never_registered", payload={"x": 1}
+            ),
+        )
+
+        self.assertEqual(worker.captured, [])
+
+    async def test_ignores_events_targeted_at_other_workers(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music",
+                target="someone_else",
+                event_name="nav_click",
+                payload={"view": "home"},
+            ),
+        )
+
+        self.assertEqual(worker.captured, [])
+        self.assertEqual(_append_frames(worker), [])
+
+    async def test_broadcast_event_with_no_target_is_handled(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target=None, event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        self.assertEqual(len(worker.captured), 1)
+
+    async def test_handler_runs_in_separate_task_so_bus_is_not_blocked(self):
+        gate = asyncio.Event()
+        observed: list[bool] = []
+
+        class _BlockingWorker(_StubUIWorker):
+            @on_ui_event("slow")
+            async def _slow(self, message):
+                await gate.wait()
+                observed.append(True)
+
+        worker = await _make_worker(cls=_BlockingWorker)
+
+        await worker.on_bus_message(
+            BusUIEventMessage(source="music", target="ui", event_name="slow", payload={})
+        )
+        self.assertEqual(observed, [])
+
+        gate.set()
+        await _settle()
+        self.assertEqual(observed, [True])
+
+    async def test_duplicate_handler_names_raise_at_init(self):
+        with self.assertRaises(ValueError):
+
+            class _Bad(UIWorker):
+                @on_ui_event("nav")
+                async def a(self, message):
+                    pass
+
+                @on_ui_event("nav")
+                async def b(self, message):
+                    pass
+
+            _Bad("ui", llm=MagicMock())
+
+    async def test_bridged_with_default_auto_inject_raises(self):
+        with self.assertRaises(ValueError) as ctx:
+            _PlainWorker("ui", llm=MagicMock(), bridged=())
+        self.assertIn("bridged", str(ctx.exception))
+        self.assertIn("auto_inject_ui_state", str(ctx.exception))
+
+    async def test_bridged_with_explicit_auto_inject_disabled_is_allowed(self):
+        worker = _PlainWorker("ui", llm=MagicMock(), bridged=(), auto_inject_ui_state=False)
+        self.assertFalse(worker._auto_inject_ui_state)
+
+    async def test_default_construction_unaffected(self):
+        worker = _PlainWorker("ui", llm=MagicMock())
+        self.assertTrue(worker._auto_inject_ui_state)
+        self.assertTrue(worker.active)
+
+
+class TestUIWorkerInjection(unittest.IsolatedAsyncioTestCase):
+    async def test_injects_xml_developer_message_by_default(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 1)
+
+        frame = frames[0]
+        self.assertFalse(frame.run_llm)
+        self.assertEqual(len(frame.messages), 1)
+        msg = frame.messages[0]
+        self.assertEqual(msg["role"], "developer")
+
+        content = msg["content"]
+        self.assertIn('<ui_event name="nav_click">', content)
+        self.assertIn("</ui_event>", content)
+        inner = content[len('<ui_event name="nav_click">') : -len("</ui_event>")]
+        self.assertEqual(json.loads(inner), {"view": "home"})
+
+    async def test_inject_events_false_disables_injection(self):
+        worker = await _make_worker(inject_events=False)
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        self.assertEqual(_append_frames(worker), [])
+        self.assertEqual(len(worker.captured), 1)
+
+    async def test_render_override_replaces_default_xml(self):
+        class _CustomRender(_StubUIWorker):
+            def render_ui_event(self, message):
+                return f"[UI] {message.event_name}"
+
+        worker = await _make_worker(cls=_CustomRender)
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 1)
+        self.assertEqual(frames[0].messages[0]["content"], "[UI] nav_click")
+
+    async def test_empty_render_skips_injection(self):
+        class _NoRender(_StubUIWorker):
+            def render_ui_event(self, message):
+                return ""
+
+        worker = await _make_worker(cls=_NoRender)
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music", target="ui", event_name="nav_click", payload={"view": "home"}
+            ),
+        )
+
+        self.assertEqual(_append_frames(worker), [])
+
+
+_SAMPLE_SNAPSHOT = {
+    "root": {
+        "ref": "e1",
+        "role": "generic",
+        "children": [
+            {
+                "ref": "e2",
+                "role": "main",
+                "children": [
+                    {"ref": "e3", "role": "heading", "name": "Home", "level": 1},
+                    {
+                        "ref": "e4",
+                        "role": "region",
+                        "name": "Trending artists",
+                        "children": [
+                            {"ref": "e5", "role": "button", "name": "Bad Bunny"},
+                            {
+                                "ref": "e6",
+                                "role": "button",
+                                "name": "Taylor Swift",
+                                "state": ["focused"],
+                            },
+                        ],
+                    },
+                ],
+            },
+        ],
+    },
+    "captured_at": 1700000000000,
+}
+
+
+class TestUIWorkerSnapshot(unittest.IsolatedAsyncioTestCase):
+    async def test_reserved_snapshot_event_stored_without_dispatch(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music",
+                target="ui",
+                event_name=_UI_SNAPSHOT_BUS_EVENT_NAME,
+                payload=_SAMPLE_SNAPSHOT,
+            ),
+        )
+
+        self.assertEqual(worker._latest_snapshot, _SAMPLE_SNAPSHOT)
+        self.assertEqual(worker.captured, [])
+        self.assertEqual(_append_frames(worker), [])
+
+    async def test_non_dict_snapshot_payload_is_ignored(self):
+        worker = await _make_worker()
+
+        await _dispatch(
+            worker,
+            BusUIEventMessage(
+                source="music",
+                target="ui",
+                event_name=_UI_SNAPSHOT_BUS_EVENT_NAME,
+                payload="not a snapshot",
+            ),
+        )
+
+        self.assertIsNone(worker._latest_snapshot)
+
+    async def test_render_ui_state_empty_without_snapshot(self):
+        worker = await _make_worker()
+        self.assertEqual(worker.render_ui_state(), "")
+
+    async def test_render_ui_state_produces_indented_block(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        rendered = worker.render_ui_state()
+
+        self.assertTrue(rendered.startswith("<ui_state>\n"))
+        self.assertTrue(rendered.endswith("\n</ui_state>"))
+
+        self.assertIn("- generic [ref=e1]:", rendered)
+        self.assertIn("- main [ref=e2]:", rendered)
+        self.assertIn('- heading "Home" [level=1] [ref=e3]', rendered)
+        self.assertIn('- region "Trending artists" [ref=e4]:', rendered)
+        self.assertIn('- button "Bad Bunny" [ref=e5]', rendered)
+        self.assertIn('- button "Taylor Swift" [focused] [ref=e6]', rendered)
+
+        self.assertIn("  - main", rendered)
+        self.assertIn("    - heading", rendered)
+        self.assertIn("      - button", rendered)
+
+    async def test_inject_ui_state_queues_expected_frame(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        await worker.inject_ui_state()
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 1)
+        frame = frames[0]
+        self.assertFalse(frame.run_llm)
+        self.assertEqual(len(frame.messages), 1)
+        msg = frame.messages[0]
+        self.assertEqual(msg["role"], "developer")
+        self.assertTrue(msg["content"].startswith("<ui_state>"))
+        self.assertTrue(msg["content"].endswith("</ui_state>"))
+
+    async def test_inject_ui_state_no_op_without_snapshot(self):
+        worker = await _make_worker()
+        await worker.inject_ui_state()
+        self.assertEqual(_append_frames(worker), [])
+
+    async def test_render_emits_grid_dims(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {
+                "ref": "e1",
+                "role": "generic",
+                "children": [
+                    {
+                        "ref": "e2",
+                        "role": "grid",
+                        "name": "Trending artists",
+                        "colcount": 8,
+                        "rowcount": 2,
+                        "children": [{"ref": "e3", "role": "button", "name": "Bad Bunny"}],
+                    },
+                ],
+            },
+            "captured_at": 1700000000000,
+        }
+
+        rendered = worker.render_ui_state()
+        self.assertIn('- grid "Trending artists" [cols=8] [rows=2] [ref=e2]', rendered)
+
+    async def test_render_preserves_offscreen_tag(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {
+                "ref": "e1",
+                "role": "generic",
+                "children": [
+                    {"ref": "e2", "role": "button", "name": "Visible"},
+                    {"ref": "e3", "role": "button", "name": "Below fold", "state": ["offscreen"]},
+                ],
+            },
+            "captured_at": 1700000000000,
+        }
+
+        rendered = worker.render_ui_state()
+        self.assertIn('- button "Visible" [ref=e2]', rendered)
+        self.assertIn('- button "Below fold" [offscreen] [ref=e3]', rendered)
+
+    async def test_visible_nodes_empty_without_snapshot(self):
+        worker = await _make_worker()
+        self.assertEqual(worker.visible_nodes(), [])
+
+    async def test_render_emits_selection_block_when_present(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {"ref": "e1", "role": "generic", "children": [{"ref": "e2", "role": "main"}]},
+            "captured_at": 1700000000000,
+            "selection": {"ref": "e2", "text": "the highlighted passage"},
+        }
+
+        rendered = worker.render_ui_state()
+        self.assertIn('<selection ref="e2">', rendered)
+        self.assertIn("the highlighted passage", rendered)
+        self.assertIn("</selection>", rendered)
+        self.assertTrue(rendered.endswith("</ui_state>"))
+        sel_idx = rendered.index('<selection ref="e2">')
+        close_idx = rendered.index("</ui_state>")
+        self.assertLess(sel_idx, close_idx)
+
+    async def test_render_omits_selection_when_missing(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {"ref": "e1", "role": "generic"},
+            "captured_at": 1700000000000,
+        }
+
+        rendered = worker.render_ui_state()
+        self.assertNotIn("<selection", rendered)
+
+    async def test_render_skips_selection_with_missing_ref_or_text(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {"ref": "e1", "role": "generic"},
+            "captured_at": 1,
+            "selection": {"ref": "e2"},
+        }
+        self.assertNotIn("<selection", worker.render_ui_state())
+
+        worker._latest_snapshot = {
+            "root": {"ref": "e1", "role": "generic"},
+            "captured_at": 1,
+            "selection": {"text": "stuff"},
+        }
+        self.assertNotIn("<selection", worker.render_ui_state())
+
+    async def test_visible_nodes_filters_offscreen_entries(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = {
+            "root": {
+                "ref": "e1",
+                "role": "generic",
+                "children": [
+                    {"ref": "e2", "role": "button", "name": "Visible"},
+                    {"ref": "e3", "role": "button", "name": "Below fold", "state": ["offscreen"]},
+                    {
+                        "ref": "e4",
+                        "role": "region",
+                        "name": "Tracks",
+                        "state": ["offscreen"],
+                        "children": [{"ref": "e5", "role": "button", "name": "Bloom"}],
+                    },
+                ],
+            },
+            "captured_at": 1700000000000,
+        }
+
+        refs = [n["ref"] for n in worker.visible_nodes()]
+        self.assertIn("e1", refs)
+        self.assertIn("e2", refs)
+        self.assertNotIn("e3", refs)
+        self.assertNotIn("e4", refs)
+        self.assertIn("e5", refs)
+
+
+class TestUIWorkerAutoInject(unittest.IsolatedAsyncioTestCase):
+    async def test_respond_auto_injects_latest_snapshot(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 2)
+        self.assertEqual(frames[0].messages[0]["role"], "developer")
+        self.assertTrue(frames[0].messages[0]["content"].startswith("<ui_state>"))
+        self.assertFalse(frames[0].run_llm)
+        self.assertEqual(frames[1].messages[0]["content"], "hi")
+        self.assertTrue(frames[1].run_llm)
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_auto_inject_ui_state_false_suppresses_injection(self):
+        worker = await _make_worker(auto_inject_ui_state=False)
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 1)
+        self.assertFalse(frames[0].messages[0]["content"].startswith("<ui_state>"))
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_auto_inject_no_op_without_snapshot(self):
+        worker = await _make_worker()
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        frames = _append_frames(worker)
+        self.assertEqual(len(frames), 1)
+        self.assertFalse(frames[0].messages[0]["content"].startswith("<ui_state>"))
+
+        await worker.respond_to_job()
+        await t
+
+
+class TestUIWorkerKeepHistory(unittest.IsolatedAsyncioTestCase):
+    async def test_default_resets_context_per_job(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        updates = _update_frames(worker)
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0].messages, [])
+        self.assertFalse(updates[0].run_llm)
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_reset_runs_before_inject(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        frame_types = [type(f).__name__ for f in worker._recorded]
+        update_idx = frame_types.index("LLMMessagesUpdateFrame")
+        append_idx = frame_types.index("LLMMessagesAppendFrame")
+        self.assertLess(update_idx, append_idx)
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_keep_history_true_skips_reset(self):
+        worker = await _make_worker(keep_history=True)
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}),
+        )
+
+        self.assertEqual(_update_frames(worker), [])
+        self.assertEqual(len(_append_frames(worker)), 2)
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_reset_context_method_emits_update_frame(self):
+        worker = await _make_worker(keep_history=True)
+
+        await worker.reset_context()
+
+        updates = _update_frames(worker)
+        self.assertEqual(len(updates), 1)
+        self.assertEqual(updates[0].messages, [])
+        self.assertFalse(updates[0].run_llm)
+
+
+class TestUIWorkerRespondToJob(unittest.IsolatedAsyncioTestCase):
+    async def test_current_job_tracks_in_flight_request(self):
+        worker = await _make_worker()
+        self.assertIsNone(worker.current_job)
+        message = BusJobRequestMessage(
+            source="voice", target="ui", job_id="t1", payload={"query": "hi"}
+        )
+        t = await _start(worker, message)
+        self.assertIs(worker.current_job, message)
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_respond_to_job_clears_current_and_sends_response(self):
+        worker = await _make_worker()
+        message = BusJobRequestMessage(source="voice", target="ui", job_id="t1")
+        t = await _start(worker, message)
+
+        await worker.respond_to_job(speak="hello")
+        await t
+
+        worker.send_job_response.assert_awaited_once()
+        call = worker.send_job_response.await_args
+        self.assertEqual(call.args[0], "t1")
+        self.assertEqual(call.kwargs["response"], {"speak": "hello"})
+        self.assertIsNone(worker.current_job)
+
+    async def test_respond_to_job_no_op_when_idle(self):
+        worker = await _make_worker()
+        await worker.respond_to_job(speak="hello")
+        worker.send_job_response.assert_not_awaited()
+
+    async def test_respond_to_job_omits_speak_when_none(self):
+        worker = await _make_worker()
+        t = await _start(worker, BusJobRequestMessage(source="voice", target="ui", job_id="t1"))
+
+        await worker.respond_to_job()
+        await t
+
+        call = worker.send_job_response.await_args
+        self.assertEqual(call.kwargs["response"], {})
+
+    async def test_respond_to_job_merges_speak_into_response(self):
+        worker = await _make_worker()
+        t = await _start(worker, BusJobRequestMessage(source="voice", target="ui", job_id="t1"))
+
+        await worker.respond_to_job({"description": "scrolled"}, speak="ok")
+        await t
+
+        call = worker.send_job_response.await_args
+        self.assertEqual(call.kwargs["response"], {"description": "scrolled", "speak": "ok"})
+
+    async def test_cancellation_frees_lock_for_subsequent_jobs(self):
+        worker = await _make_worker()
+        msg_a = _respond_msg("a")
+        msg_b = _respond_msg("b")
+
+        await worker._handle_job_request(msg_a)
+        await _settle()
+        self.assertIs(worker.current_job, msg_a)
+        self.assertTrue(worker._job_locks["respond"].locked())
+
+        await worker._handle_job_cancel(
+            BusJobCancelMessage(source="voice", target="ui", job_id="a")
+        )
+        await _settle()
+        self.assertIsNone(worker.current_job)
+        self.assertFalse(worker._job_locks["respond"].locked())
+
+        await worker._handle_job_request(msg_b)
+        await _settle()
+        self.assertIs(worker.current_job, msg_b)
+
+        await worker.respond_to_job(speak="B done")
+        await _settle()
+
+    async def test_cancel_unknown_job_id_is_noop(self):
+        worker = await _make_worker()
+        msg_a = _respond_msg("a")
+
+        await worker._handle_job_request(msg_a)
+        await _settle()
+
+        await worker._handle_job_cancel(
+            BusJobCancelMessage(source="voice", target="ui", job_id="unrelated")
+        )
+        await _settle()
+
+        self.assertIs(worker.current_job, msg_a)
+        self.assertTrue(worker._job_locks["respond"].locked())
+
+        await worker.respond_to_job(speak="A done")
+        await _settle()
+
+    async def test_concurrent_same_name_jobs_serialize(self):
+        worker = await _make_worker()
+        msg_a = _respond_msg("a")
+        msg_b = _respond_msg("b")
+
+        await worker._handle_job_request(msg_a)
+        await _settle()
+        self.assertIs(worker.current_job, msg_a)
+
+        await worker._handle_job_request(msg_b)
+        await _settle()
+        self.assertIs(worker.current_job, msg_a)
+
+        await worker.respond_to_job(speak="A done")
+        await _settle()
+        self.assertIs(worker.current_job, msg_b)
+
+        await worker.respond_to_job(speak="B done")
+        await _settle()
+        self.assertIsNone(worker.current_job)
+
+
+class TestUIWorkerRespondJob(unittest.IsolatedAsyncioTestCase):
+    async def test_respond_handler_runs_after_setup(self):
+        worker = await _make_worker()
+        worker._latest_snapshot = _SAMPLE_SNAPSHOT
+
+        t = await _start(
+            worker,
+            BusJobRequestMessage(
+                source="voice", target="ui", job_id="t1", payload={"query": "hello"}
+            ),
+        )
+
+        appends = _append_frames(worker)
+        self.assertEqual(len(appends), 2)
+        self.assertTrue(appends[0].messages[0]["content"].startswith("<ui_state>"))
+        self.assertFalse(appends[0].run_llm)
+        self.assertEqual(appends[1].messages[0]["content"], "hello")
+        self.assertTrue(appends[1].run_llm)
+        self.assertEqual(worker.current_job.job_id, "t1")
+
+        await worker.respond_to_job(speak="done")
+        await t
+        self.assertIsNone(worker.current_job)
+        worker.send_job_response.assert_awaited_once()
+
+    async def test_render_query_override(self):
+        class _Custom(_StubUIWorker):
+            def render_query(self, message):
+                return f"Q: {message.payload['q']}"
+
+        worker = await _make_worker(cls=_Custom)
+        t = await _start(
+            worker,
+            BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"q": "hi"}),
+        )
+
+        query_frames = [f for f in _append_frames(worker) if f.run_llm]
+        self.assertEqual(query_frames[0].messages[0]["content"], "Q: hi")
+
+        await worker.respond_to_job()
+        await t
+
+    async def test_handler_failure_clears_state(self):
+        class _Boom(_StubUIWorker):
+            def render_query(self, message):
+                raise RuntimeError("boom")
+
+        worker = await _make_worker(cls=_Boom)
+
+        with self.assertRaises(RuntimeError):
+            await worker.respond_with_llm(
+                BusJobRequestMessage(
+                    source="voice", target="ui", job_id="t1", payload={"query": "x"}
+                )
+            )
+
+        self.assertIsNone(worker.current_job)
+        self.assertIsNone(worker._pending)
+
+
+class TestUIWorkerPromptGuide(unittest.IsolatedAsyncioTestCase):
+    async def test_default_appends_ui_state_prompt_guide(self):
+        worker = UIWorker("ui", llm=MagicMock(), active=False)
+        worker.llm.append_system_instruction.assert_called_once_with(UI_STATE_PROMPT_GUIDE)
+
+    async def test_custom_prompt_guide_overrides(self):
+        worker = UIWorker("ui", llm=MagicMock(), active=False, prompt_guide="MY GUIDE")
+        worker.llm.append_system_instruction.assert_called_once_with("MY GUIDE")
+
+    async def test_none_disables_injection(self):
+        worker = UIWorker("ui", llm=MagicMock(), active=False, prompt_guide=None)
+        worker.llm.append_system_instruction.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()