Mark realtime LLM services with RealtimeServiceInfo + emit metadata at start

Realtime (speech-to-speech) LLM services need to advertise themselves to
the rest of the pipeline so downstream components can adapt. Add a new
RealtimeServiceMetadataFrame subtype of ServiceMetadataFrame, following
the STTMetadataFrame precedent.

LLMService gains a single ClassVar, _realtime_service_info, typed
RealtimeServiceInfo | None and defaulting to None. The presence of a
populated instance is what marks a service as realtime, and the
RealtimeServiceInfo dataclass carries the per-service knobs the rest of
the pipeline needs — currently just emits_user_turn_frames. Keeping it
all under one optional ClassVar avoids stranding realtime-only knobs on
the generic LLMService surface; non-realtime services keep the default
None and the realtime-specific machinery stays inert.

When _realtime_service_info is set, the base service auto-broadcasts
RealtimeServiceMetadataFrame right after StartFrame propagates downstream
(same ordering as STT). When emits_user_turn_frames is False, a one-time
INFO log at start explains which pipeline processors depend on those
frames (RTVI client speech events, TurnTrackingObserver,
AudioBufferProcessor turn recording, UserIdleController, user mute
strategies, voicemail detector) and how to add local VAD if needed.

Set the ClassVar on the seven realtime services: OpenAI Realtime, Azure
Realtime (via inheritance), Inworld, Grok/xAI Realtime all emit
user-turn frames; Gemini Live (and Gemini Live Vertex via inheritance),
AWS Nova Sonic, Ultravox do not.

In a follow-up commit, LLMContextAggregatorPair will consume
RealtimeServiceMetadataFrame to surface a one-time recommendation when
realtime_service_mode is not configured.

src/pipecat/frames/frames.py

@@ -1439,6 +1439,27 @@ class STTMetadataFrame(ServiceMetadataFrame):
     ttfs_p99_latency: float
 
 
+@dataclass
+class RealtimeServiceMetadataFrame(ServiceMetadataFrame):
+    """Metadata announcing a realtime (speech-to-speech) LLM service.
+
+    Broadcast by realtime LLM services at pipeline start so downstream
+    processors — notably ``LLMContextAggregatorPair`` — can detect that
+    a realtime service is in the pipeline. The aggregator uses this to
+    surface a one-time recommendation to opt in to
+    ``RealtimeServiceModeConfig`` when it hasn't been configured.
+
+    Parameters:
+        emits_user_turn_frames: Whether this service emits
+            ``UserStartedSpeakingFrame`` / ``UserStoppedSpeakingFrame``
+            from server-side turn signals. False for services with no
+            server-side turn signals (e.g. Gemini Live, AWS Nova Sonic,
+            Ultravox).
+    """
+
+    emits_user_turn_frames: bool = True
+
+
 @dataclass
 class ServiceSwitcherRequestMetadataFrame(ControlFrame):
     """Request a service to re-emit its metadata frames.

src/pipecat/services/aws/nova_sonic/llm.py

@@ -56,7 +56,7 @@ from pipecat.services.aws.nova_sonic.session_continuation import (
     SessionContinuationHelper,
     SessionContinuationParams,
 )
-from pipecat.services.llm_service import LLMService
+from pipecat.services.llm_service import LLMService, RealtimeServiceInfo
 from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given
 from pipecat.utils.time import time_now_iso8601
 
@@ -249,6 +249,10 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
     # Override the default adapter to use the AWSNovaSonicLLMAdapter one
     adapter_class = AWSNovaSonicLLMAdapter
 
+    # Realtime (speech-to-speech) service. Does NOT emit
+    # UserStarted/StoppedSpeakingFrame from server-side turn signals.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False)
+
     def __init__(
         self,
         *,

src/pipecat/services/google/gemini_live/llm.py

@@ -62,7 +62,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMe
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult
 from pipecat.services.google.utils import update_google_client_http_options
-from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo
 from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given
 from pipecat.transcriptions.language import Language, resolve_language
 from pipecat.utils.string import match_endofsentence
@@ -369,6 +369,11 @@ class GeminiLiveLLMService(LLMService[GeminiLLMAdapter]):
     # Overriding the default adapter to use the Gemini one.
     adapter_class = GeminiLLMAdapter
 
+    # Realtime (speech-to-speech) service. Does NOT emit
+    # UserStarted/StoppedSpeakingFrame from server-side turn signals —
+    # the API exposes an `interrupted` event but no turn-start/-end.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False)
+
     @property
     def _is_gemini_3(self) -> bool:
         """Check if the current model is a Gemini 3.x model."""

src/pipecat/services/inworld/realtime/llm.py

@@ -51,7 +51,7 @@ from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators import async_tool_messages
 from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo
 from pipecat.services.settings import (
     NOT_GIVEN,
     LLMSettings,
@@ -245,6 +245,10 @@ class InworldRealtimeLLMService(LLMService[InworldRealtimeLLMAdapter]):
 
     adapter_class = InworldRealtimeLLMAdapter
 
+    # Realtime (speech-to-speech) service. Emits UserStarted/Stopped
+    # speaking frames from server-side VAD events.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True)
+
     # Target ~60ms audio chunks when sending to Inworld (16-bit mono).
     _AUDIO_CHUNK_TARGET_MS = 60
 

src/pipecat/services/llm_service.py

@@ -16,6 +16,7 @@ from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
 from typing import (
     Any,
+    ClassVar,
     Generic,
     Protocol,
     cast,
@@ -48,6 +49,7 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMTextFrame,
     LLMUpdateSettingsFrame,
+    RealtimeServiceMetadataFrame,
     StartFrame,
 )
 from pipecat.processors.aggregators.llm_context import (
@@ -97,6 +99,31 @@ class FunctionCallResultCallback(Protocol):
         ...
 
 
+@dataclass(frozen=True)
+class RealtimeServiceInfo:
+    """Per-service metadata for realtime (speech-to-speech) LLM services.
+
+    Realtime LLM subclasses set ``LLMService._realtime_service_info`` to a
+    populated instance; the presence of a non-None value is what marks a
+    service as realtime. Non-realtime services keep the default ``None``.
+
+    Carries the configuration ``LLMService`` and
+    ``LLMContextAggregatorPair`` need to wire up realtime behavior:
+    auto-broadcasting ``RealtimeServiceMetadataFrame`` at start, the
+    startup INFO log for services with no server-side turn signals, and
+    the aggregator's one-time recommendation log.
+
+    Parameters:
+        emits_user_turn_frames: Whether the service emits
+            ``UserStartedSpeakingFrame`` / ``UserStoppedSpeakingFrame``
+            from server-side turn signals. False for services with no
+            server-side turn signals (e.g. Gemini Live, AWS Nova Sonic,
+            Ultravox).
+    """
+
+    emits_user_turn_frames: bool = True
+
+
 @dataclass
 class FunctionCallParams:
     """Parameters for a function call.
@@ -244,6 +271,15 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter]
     # However, subclasses should override this with a more specific adapter when necessary.
     adapter_class: type[BaseLLMAdapter] = OpenAILLMAdapter
 
+    # Marker + per-service config for realtime (speech-to-speech) LLM
+    # services. Realtime subclasses override this with a populated
+    # ``RealtimeServiceInfo`` instance — the presence of a non-None value
+    # is what marks the service as realtime. Non-realtime services keep
+    # the default ``None`` and the realtime-specific machinery
+    # (auto-broadcast of ``RealtimeServiceMetadataFrame``, startup INFO
+    # log for services without server-side turn signals) stays inert.
+    _realtime_service_info: ClassVar[RealtimeServiceInfo | None] = None
+
     # Returned to the LLM as the tool result when an unavailable function is
     # called. Deliberately neutral about future availability so the LLM can
     # pick the function up again if it returns (e.g. via the
@@ -363,6 +399,21 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter]
             await self._create_sequential_runner_task()
         if self._enable_async_tool_cancellation and self._has_async_tools():
             self._setup_async_tool_cancellation()
+        if (
+            self._realtime_service_info is not None
+            and not self._realtime_service_info.emits_user_turn_frames
+        ):
+            logger.info(
+                f"{self} does not emit UserStartedSpeakingFrame/"
+                "UserStoppedSpeakingFrame. Pipeline processors that depend on "
+                "these frames (RTVI client speech events, TurnTrackingObserver, "
+                "AudioBufferProcessor turn recording, UserIdleController, user "
+                "mute strategies, voicemail detector) will not activate. To "
+                "produce them locally, add `vad_analyzer=` to "
+                "LLMUserAggregatorParams. Note: local turn detection is a "
+                "heuristic; its boundaries may not match the provider's actual "
+                "server-side turn decisions and can desynchronize in subtle ways."
+            )
 
     async def stop(self, frame: EndFrame):
         """Stop the LLM service.
@@ -495,6 +546,23 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService, Generic[TAdapter]
 
         await super().push_frame(frame, direction)
 
+        # Broadcast realtime-service metadata immediately after the
+        # StartFrame propagates downstream, mirroring the order STT
+        # services use for STTMetadataFrame. The aggregator (upstream)
+        # already received its own StartFrame and is ready to process
+        # the broadcast; downstream processors see StartFrame then the
+        # metadata in their queues.
+        if (
+            self._realtime_service_info is not None
+            and isinstance(frame, StartFrame)
+            and direction == FrameDirection.DOWNSTREAM
+        ):
+            await self.broadcast_frame(
+                RealtimeServiceMetadataFrame,
+                service_name=self.name,
+                emits_user_turn_frames=self._realtime_service_info.emits_user_turn_frames,
+            )
+
     async def _push_llm_text(self, text: str):
         """Push LLM text, using turn completion detection if enabled.
 

src/pipecat/services/openai/realtime/llm.py

@@ -51,7 +51,7 @@ from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators import async_tool_messages
 from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo
 from pipecat.services.openai._constants import OPENAI_REALTIME_WHISPER_MODEL, OPENAI_SAMPLE_RATE
 from pipecat.services.settings import (
     NOT_GIVEN,
@@ -212,6 +212,10 @@ class OpenAIRealtimeLLMService(LLMService[OpenAIRealtimeLLMAdapter]):
     # Overriding the default adapter to use the OpenAIRealtimeLLMAdapter one.
     adapter_class = OpenAIRealtimeLLMAdapter
 
+    # Realtime (speech-to-speech) service. Emits UserStarted/Stopped
+    # speaking frames from server-side VAD events.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True)
+
     def __init__(
         self,
         *,

src/pipecat/services/ultravox/llm.py

@@ -48,7 +48,7 @@ from pipecat.frames.frames import (
 from pipecat.processors.aggregators import async_tool_messages
 from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo
 from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, assert_given
 from pipecat.utils.time import time_now_iso8601
 
@@ -179,6 +179,10 @@ class UltravoxRealtimeLLMService(LLMService):
     Settings = UltravoxRealtimeLLMSettings
     _settings: Settings
 
+    # Realtime (speech-to-speech) service. Does NOT emit
+    # UserStarted/StoppedSpeakingFrame from server-side turn signals.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=False)
+
     def __init__(
         self,
         *,

src/pipecat/services/xai/realtime/llm.py

@@ -50,7 +50,7 @@ from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators import async_tool_messages
 from pipecat.processors.aggregators.llm_context import LLMContext, LLMSpecificMessage
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService, RealtimeServiceInfo
 from pipecat.services.settings import (
     NOT_GIVEN,
     LLMSettings,
@@ -203,6 +203,10 @@ class GrokRealtimeLLMService(LLMService[GrokRealtimeLLMAdapter]):
     # Use the Grok-specific adapter
     adapter_class = GrokRealtimeLLMAdapter
 
+    # Realtime (speech-to-speech) service. Emits UserStarted/Stopped
+    # speaking frames from server-side VAD events.
+    _realtime_service_info = RealtimeServiceInfo(emits_user_turn_frames=True)
+
     def __init__(
         self,
         *,