fix: fix speech to speech model transitions (#545)

* fix: fix transition logic for realtime providers

* chore: run formatter

* chore: generate SDK and fix other realtime providers

* fix: fix ultravox node transitions

api/services/pipecat/realtime/azure_realtime.py

@@ -1,9 +1,9 @@
 """Dograh subclass of pipecat's Azure OpenAI Realtime LLM service.
 
 Layers Dograh engine integration quirks (mute gating, TTSSpeakFrame greeting
-trigger, LLMMessagesAppendFrame handling, deferred tool calls) onto pipecat's
-AzureRealtimeLLMService, mirroring what DograhOpenAIRealtimeLLMService does
-for the standard OpenAI Realtime endpoint.
+trigger, LLMMessagesAppendFrame handling, workflow-control deferral) onto
+pipecat's AzureRealtimeLLMService, mirroring what
+DograhOpenAIRealtimeLLMService does for the standard OpenAI Realtime endpoint.
 """
 
 import json
@@ -40,7 +40,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
     - User-mute audio gating
     - TTSSpeakFrame as initial-response trigger
     - One-off LLMMessagesAppendFrame handling
-    - Deferred tool calls until bot finishes speaking
+    - Workflow-control calls deferred until bot finishes speaking
     - finalized=True on TranscriptionFrame for consistency
     """
 
@@ -49,7 +49,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
         self._user_is_muted: bool = False
         self._handled_initial_context: bool = False
         self._bot_is_speaking: bool = False
-        self._deferred_function_calls: list[FunctionCallFromLLM] = []
+        self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
         self._pending_initial_greeting_text: str | None = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -81,7 +81,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
             self._bot_is_speaking = True
         elif isinstance(frame, BotStoppedSpeakingFrame):
             self._bot_is_speaking = False
-            await self._run_pending_function_calls()
+            await self._run_pending_node_transition_function_calls()
         await super().process_frame(frame, direction)
 
     async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@@ -247,18 +247,19 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
             )
         )
 
-    async def _run_pending_function_calls(self):
-        if not self._deferred_function_calls:
+    async def _run_pending_node_transition_function_calls(self):
+        if not self._deferred_node_transition_function_calls:
             return
-        function_calls = self._deferred_function_calls
-        self._deferred_function_calls = []
+        function_calls = self._deferred_node_transition_function_calls
+        self._deferred_node_transition_function_calls = []
         logger.debug(
-            f"{self}: executing {len(function_calls)} deferred function call(s) "
-            "after bot turn ended"
+            f"{self}: executing {len(function_calls)} deferred workflow-control "
+            "call(s) after bot turn ended"
         )
         await self.run_function_calls(function_calls)
 
     async def _handle_evt_function_call_arguments_done(self, evt):
+        """Run ordinary tools immediately and defer workflow-control calls."""
         try:
             args = json.loads(evt.arguments)
 
@@ -275,10 +276,14 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
                     )
                 ]
 
-                if self._bot_is_speaking:
-                    self._deferred_function_calls.extend(function_calls)
+                is_node_transition = self._function_is_node_transition(
+                    function_call_item.name
+                )
+                if self._bot_is_speaking and is_node_transition:
+                    self._deferred_node_transition_function_calls.extend(function_calls)
                     logger.debug(
-                        f"{self}: deferring function call {function_call_item.name} "
+                        f"{self}: deferring workflow-control call "
+                        f"{function_call_item.name} "
                         "until bot stops speaking"
                     )
                 else:

api/services/pipecat/realtime/gemini_live.py

@@ -9,8 +9,8 @@ Layers Dograh engine integration quirks onto upstream-pristine
 - **Reconnect on node transitions.** Gemini Live cannot update
   ``system_instruction`` mid-session, so a setting change triggers a
   reconnect (deferred until the bot turn ends if currently responding).
-- **Function-call deferral.** Tool calls emitted mid-turn are queued and run
-  when the bot stops speaking, to avoid racing the turn's audio.
+- **Node-transition deferral.** Node-transition calls emitted mid-turn are
+  queued and run when the bot stops speaking, to avoid cutting off its audio.
 - **User-mute audio gating.** ``UserMuteStarted/StoppedFrame`` from the
   user aggregator gates whether incoming audio is forwarded to Gemini.
 - **TTSSpeakFrame as greeting trigger.** The engine queues a TTSSpeakFrame
@@ -18,6 +18,7 @@ Layers Dograh engine integration quirks onto upstream-pristine
   it and runs the initial-context path.
 """
 
+import asyncio
 from typing import Any
 
 from google.genai.types import Content, Part
@@ -44,6 +45,11 @@ from pipecat.utils.tracing.service_decorators import traced_gemini_live
 class DograhGeminiLiveLLMService(GeminiLiveLLMService):
     """Gemini Live with Dograh engine integration quirks. See module docstring."""
 
+    # Gemini input transcription is delivered independently from tool calls.
+    # Give late transcription messages a small window to arrive before running
+    # a node-transition function and tearing down the current Live connection.
+    _NODE_TRANSITION_TRANSCRIPTION_GRACE_SECONDS = 0.5
+
     # Route tool schemas through Gemini's ``parameters_json_schema`` field so
     # MCP/imported tools that use JSON Schema keywords (``const``, ``not``,
     # nested ``anyOf``) rejected by the strict ``Schema`` model are accepted.
@@ -59,15 +65,19 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
         # Guards initial-response triggering against double-firing across the
         # initial TTSSpeakFrame and any LLMContextFrame that may arrive.
         self._handled_initial_context: bool = False
-        # When a system_instruction change arrives mid-bot-turn, the reconnect
-        # is queued and drained when the turn ends.
-        self._reconnect_pending: bool = False
-        # Function calls emitted by Gemini mid-bot-turn are deferred here and
-        # invoked when the turn ends, so they don't race the turn's audio.
-        self._pending_function_calls: list[FunctionCallFromLLM] = []
+        # Node-transition calls emitted mid-bot-turn are deferred here so the
+        # transition does not tear down Gemini while it is still producing audio.
+        self._pending_node_transition_function_calls: list[FunctionCallFromLLM] = []
         # Text greeting captured from the first TTSSpeakFrame while the Gemini
         # session is still connecting.
         self._pending_initial_greeting_text: str | None = None
+        self._transition_function_call_task: asyncio.Task | None = None
+        # Intentional node changes use a fresh, context-seeded connection rather
+        # than a potentially stale session-resumption handle. The new connection
+        # remains gated until the function-call result has landed in LLMContext.
+        self._awaiting_node_transition_context: bool = False
+        self._node_transition_context_received: bool = False
+        self._node_transition_context_seed_started: bool = False
 
     # ------------------------------------------------------------------
     # Hooks from upstream GeminiLiveLLMService
@@ -78,33 +88,109 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
         # lets pre-call fetch populate template variables first.
         return bool(self._settings.system_instruction)
 
+    def _requires_node_transition_context_aggregation(self) -> bool:
+        # A node transition replaces the current Gemini Live connection and
+        # seeds the new one from our local LLMContext. Wait for the upstream
+        # user aggregator to commit any final TranscriptionFrame before
+        # set_node() changes the prompt and starts that reconnect.
+        return True
+
+    async def cleanup(self) -> None:
+        """Cancel a delayed transition before tearing down the Live session."""
+        if self._transition_function_call_task:
+            await self.cancel_task(self._transition_function_call_task)
+            self._transition_function_call_task = None
+        await super().cleanup()
+
     async def _handle_changed_settings(self, changed: dict[str, Any]) -> set[str]:
         if "system_instruction" not in changed:
             return set()
+
+        # PipecatEngine updates system_instruction only from set_node(). The
+        # first set_node happens before a Live session exists; every later one
+        # is a node transition whose tool call has already been deferred until
+        # the current bot turn finishes.
         if not self._session:
             # First-time setting after deferred-connect.
             await self._connect()
-        elif self._bot_is_responding:
-            # Bot is mid-turn — drain the reconnect when it ends so we don't
-            # cut the bot off mid-utterance.
-            self._reconnect_pending = True
         else:
-            await self._reconnect()
+            await self._reconnect_for_node_transition()
         return {"system_instruction"}
 
     async def _run_or_defer_function_calls(
         self, function_calls_llm: list[FunctionCallFromLLM]
     ):
+        if not self._contains_node_transition(function_calls_llm):
+            await super()._run_or_defer_function_calls(function_calls_llm)
+            return
+
+        # Keep a provider tool-call batch together. Splitting a mixed batch here
+        # would discard Pipecat's shared function-call group and could trigger an
+        # LLM run before every result from the original batch has arrived.
         if self._bot_is_responding:
             # Latest batch wins; Gemini emits tool calls as one batch per
             # tool_call message, so this overwrite is intentional.
-            self._pending_function_calls = function_calls_llm
+            self._pending_node_transition_function_calls = function_calls_llm
             logger.debug(
-                f"{self}: deferring {len(function_calls_llm)} function call(s) "
+                f"{self}: deferring {len(function_calls_llm)} node-transition "
+                "function call(s) "
                 "until bot turn ends"
             )
             return
-        await super()._run_or_defer_function_calls(function_calls_llm)
+
+        self._schedule_node_transition_function_calls(function_calls_llm)
+
+    def _contains_node_transition(
+        self, function_calls_llm: list[FunctionCallFromLLM]
+    ) -> bool:
+        return any(self._is_node_transition(fc) for fc in function_calls_llm)
+
+    def _is_node_transition(self, function_call: FunctionCallFromLLM) -> bool:
+        return self._function_is_node_transition(function_call.function_name)
+
+    def _schedule_node_transition_function_calls(
+        self, function_calls_llm: list[FunctionCallFromLLM]
+    ) -> None:
+        """Run transition calls after late input transcription has settled."""
+        if (
+            self._transition_function_call_task
+            and not self._transition_function_call_task.done()
+        ):
+            logger.warning(
+                f"{self}: node-transition function call already pending; "
+                "ignoring duplicate batch"
+            )
+            return
+
+        async def _run_after_transcription_grace() -> None:
+            try:
+                await asyncio.sleep(self._NODE_TRANSITION_TRANSCRIPTION_GRACE_SECONDS)
+                await self._flush_pending_user_transcription()
+                await self.run_function_calls(function_calls_llm)
+            finally:
+                self._transition_function_call_task = None
+
+        self._transition_function_call_task = self.create_task(
+            _run_after_transcription_grace(),
+            name=f"{self}::node-transition-function-calls",
+        )
+
+    async def _flush_pending_user_transcription(self) -> None:
+        """Publish any punctuationless user transcript before a node handoff."""
+        if self._transcription_timeout_task:
+            if not self._transcription_timeout_task.done():
+                await self.cancel_task(self._transcription_timeout_task)
+            self._transcription_timeout_task = None
+
+        if not self._user_transcription_buffer:
+            return
+
+        text = self._user_transcription_buffer
+        self._user_transcription_buffer = ""
+        logger.debug(
+            f"{self}: flushing pending user transcription before node transition"
+        )
+        await self._push_user_transcription(text, result=None)
 
     # ------------------------------------------------------------------
     # State-transition side effects
@@ -114,22 +200,35 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
         was_responding = self._bot_is_responding
         await super()._set_bot_is_responding(responding)
         if was_responding and not responding:
-            await self._run_pending_function_calls()
-            if self._reconnect_pending:
-                self._reconnect_pending = False
-                await self._reconnect()
+            await self._run_pending_node_transition_function_calls()
 
-    async def _run_pending_function_calls(self):
-        """Run any function calls deferred during the bot's last turn."""
-        if not self._pending_function_calls:
+    async def _run_pending_node_transition_function_calls(self):
+        """Run any node-transition calls deferred during the bot's last turn."""
+        if not self._pending_node_transition_function_calls:
             return
-        fcs = self._pending_function_calls
-        self._pending_function_calls = []
+        fcs = self._pending_node_transition_function_calls
+        self._pending_node_transition_function_calls = []
         logger.debug(
-            f"{self}: executing {len(fcs)} deferred function call(s) "
+            f"{self}: executing {len(fcs)} deferred node-transition call(s) "
             "after bot turn ended"
         )
-        await self.run_function_calls(fcs)
+        self._schedule_node_transition_function_calls(fcs)
+
+    async def _reconnect_for_node_transition(self) -> None:
+        """Start a fresh connection and wait to seed the completed context.
+
+        Gemini can report ``resumable=False`` while generating or executing a
+        function call. A workflow transition happens at exactly that boundary,
+        so using the last (older) resumption handle can omit the triggering user
+        turn. Use the local LLMContext as the source of truth for this intentional
+        handoff instead.
+        """
+        self._awaiting_node_transition_context = True
+        self._node_transition_context_received = False
+        self._node_transition_context_seed_started = False
+        self._session_resumption_handle = None
+        await self._disconnect()
+        await self._connect(session_resumption_handle=None)
 
     # ------------------------------------------------------------------
     # Frame handling: mute, TTSSpeakFrame, BotStoppedSpeakingFrame flush
@@ -165,9 +264,9 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
         if isinstance(frame, BotStoppedSpeakingFrame):
             # Belt-and-suspenders: the main drain happens in
             # _set_bot_is_responding(False), but if Gemini delays turn_complete
-            # past the audible end of the turn, flushing here ensures pending
-            # function calls fire promptly.
-            await self._run_pending_function_calls()
+            # past the audible end of the turn, flushing here ensures a pending
+            # node transition fires promptly.
+            await self._run_pending_node_transition_function_calls()
             # Fall through to super for the actual push.
         await super().process_frame(frame, direction)
 
@@ -185,6 +284,11 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
     # ------------------------------------------------------------------
 
     async def _handle_context(self, context: LLMContext):
+        if self._awaiting_node_transition_context:
+            self._context = context
+            self._node_transition_context_received = True
+            await self._maybe_seed_node_transition_context()
+            return
         if not self._handled_initial_context:
             self._handled_initial_context = True
             self._context = context
@@ -249,6 +353,12 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
             f"In _handle_session_ready self._run_llm_when_session_ready: {self._run_llm_when_session_ready}"
         )
         self._session = session
+        if self._awaiting_node_transition_context:
+            # Do not accept realtime input until the function-call result frame
+            # has updated the shared context and that complete history is seeded.
+            self._ready_for_realtime_input = False
+            await self._maybe_seed_node_transition_context()
+            return
         self._ready_for_realtime_input = True
         if self._run_llm_when_session_ready:
             # Context arrived before session was ready — fulfil the queued
@@ -266,3 +376,25 @@ class DograhGeminiLiveLLMService(GeminiLiveLLMService):
         # a handle (e.g. node transitions before any handle was issued) are
         # followed by a function-call-result LLMContextFrame which feeds the
         # updated-context branch in _handle_context.
+
+    async def _maybe_seed_node_transition_context(self) -> None:
+        if (
+            not self._awaiting_node_transition_context
+            or not self._node_transition_context_received
+            or not self._session
+            or self._node_transition_context_seed_started
+        ):
+            return
+
+        self._node_transition_context_seed_started = True
+        try:
+            # The complete tool result is already present in the history being
+            # seeded, so mark it delivered locally instead of sending a provider
+            # tool response for a call that the fresh session never issued.
+            await self._process_completed_function_calls(send_new_results=False)
+            await self._create_initial_response()
+            self._awaiting_node_transition_context = False
+            self._node_transition_context_received = False
+            await self._drain_pending_tool_results()
+        finally:
+            self._node_transition_context_seed_started = False

api/services/pipecat/realtime/grok_realtime.py

@@ -11,8 +11,9 @@ Adds:
   flow kicks off the bot's first response.
 - **One-off LLMMessagesAppendFrame handling** for ephemeral realtime prompts
   like user-idle checks, without mutating Dograh's local ``LLMContext``.
-- **Function-call deferral** until the bot finishes speaking, to avoid racing
-  tool execution with the active audio turn.
+- **Workflow-control deferral** so node transitions, call termination, and
+  transfers wait for any current bot audio to finish while ordinary tools run
+  immediately.
 - **finalized=True on TranscriptionFrame** for parity with Dograh's other
   realtime providers.
 """
@@ -50,7 +51,7 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
         self._user_is_muted: bool = False
         self._handled_initial_context: bool = False
         self._bot_is_speaking: bool = False
-        self._deferred_function_calls: list[FunctionCallFromLLM] = []
+        self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
         self._pending_initial_greeting_text: str | None = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -82,7 +83,7 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
             self._bot_is_speaking = True
         elif isinstance(frame, BotStoppedSpeakingFrame):
             self._bot_is_speaking = False
-            await self._run_pending_function_calls()
+            await self._run_pending_node_transition_function_calls()
         await super().process_frame(frame, direction)
 
     async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@@ -251,19 +252,19 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
             )
         )
 
-    async def _run_pending_function_calls(self):
-        if not self._deferred_function_calls:
+    async def _run_pending_node_transition_function_calls(self):
+        if not self._deferred_node_transition_function_calls:
             return
-        function_calls = self._deferred_function_calls
-        self._deferred_function_calls = []
+        function_calls = self._deferred_node_transition_function_calls
+        self._deferred_node_transition_function_calls = []
         logger.debug(
-            f"{self}: executing {len(function_calls)} deferred function call(s) "
-            "after bot turn ended"
+            f"{self}: executing {len(function_calls)} deferred workflow-control "
+            "call(s) after bot turn ended"
         )
         await self.run_function_calls(function_calls)
 
     async def _handle_evt_function_call_arguments_done(self, evt):
-        """Process or defer tool calls until the bot finishes speaking."""
+        """Run ordinary tools immediately and defer workflow-control calls."""
         try:
             args = json.loads(evt.arguments)
 
@@ -281,10 +282,11 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
                     )
                 ]
 
-                if self._bot_is_speaking:
-                    self._deferred_function_calls.extend(function_calls)
+                is_node_transition = self._function_is_node_transition(function_name)
+                if self._bot_is_speaking and is_node_transition:
+                    self._deferred_node_transition_function_calls.extend(function_calls)
                     logger.debug(
-                        f"{self}: deferring function call {function_name} "
+                        f"{self}: deferring workflow-control call {function_name} "
                         "until bot stops speaking"
                     )
                 else:

api/services/pipecat/realtime/openai_realtime.py

@@ -3,8 +3,8 @@
 Layers Dograh engine integration quirks onto upstream-pristine
 :class:`OpenAIRealtimeLLMService`. Substantially smaller than the Gemini
 subclass because OpenAI Realtime supports runtime ``session.update`` for
-both ``system_instruction`` and tools — no reconnect/defer-tool-call
-machinery needed.
+both ``system_instruction`` and tools, so node changes do not require a
+reconnect.
 
 Adds:
 
@@ -13,6 +13,9 @@ Adds:
   flow kicks off the bot's first response.
 - **One-off LLMMessagesAppendFrame handling** for ephemeral realtime prompts
   like user-idle checks, without mutating Dograh's local ``LLMContext``.
+- **Workflow-control deferral** so node transitions, call termination, and
+  transfers wait for any current bot audio to finish while ordinary tools run
+  immediately.
 - **finalized=True on TranscriptionFrame** because every OpenAI
   transcription via the ``completed`` event is final by construction.
 """
@@ -53,10 +56,10 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
         # LLMContextFrame arrives, so upstream's "first arrival means
         # self._context is None" check no longer works.
         self._handled_initial_context: bool = False
-        # Track bot speech locally so tool calls can be deferred until the bot
-        # has finished speaking, matching Dograh's Gemini Live behavior.
+        # Track bot speech locally so workflow-control calls can wait until the
+        # bot has finished speaking without delaying ordinary tools.
         self._bot_is_speaking: bool = False
-        self._deferred_function_calls: list[FunctionCallFromLLM] = []
+        self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
         self._pending_initial_greeting_text: str | None = None
 
     # ------------------------------------------------------------------
@@ -100,7 +103,7 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
             self._bot_is_speaking = True
         elif isinstance(frame, BotStoppedSpeakingFrame):
             self._bot_is_speaking = False
-            await self._run_pending_function_calls()
+            await self._run_pending_node_transition_function_calls()
         await super().process_frame(frame, direction)
 
     async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@@ -268,19 +271,19 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
             )
         )
 
-    async def _run_pending_function_calls(self):
-        if not self._deferred_function_calls:
+    async def _run_pending_node_transition_function_calls(self):
+        if not self._deferred_node_transition_function_calls:
             return
-        function_calls = self._deferred_function_calls
-        self._deferred_function_calls = []
+        function_calls = self._deferred_node_transition_function_calls
+        self._deferred_node_transition_function_calls = []
         logger.debug(
-            f"{self}: executing {len(function_calls)} deferred function call(s) "
-            "after bot turn ended"
+            f"{self}: executing {len(function_calls)} deferred workflow-control "
+            "call(s) after bot turn ended"
         )
         await self.run_function_calls(function_calls)
 
     async def _handle_evt_function_call_arguments_done(self, evt):
-        """Process or defer tool calls until the bot finishes speaking."""
+        """Run ordinary tools immediately and defer workflow-control calls."""
         try:
             args = json.loads(evt.arguments)
 
@@ -297,10 +300,14 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
                     )
                 ]
 
-                if self._bot_is_speaking:
-                    self._deferred_function_calls.extend(function_calls)
+                is_node_transition = self._function_is_node_transition(
+                    function_call_item.name
+                )
+                if self._bot_is_speaking and is_node_transition:
+                    self._deferred_node_transition_function_calls.extend(function_calls)
                     logger.debug(
-                        f"{self}: deferring function call {function_call_item.name} "
+                        f"{self}: deferring workflow-control call "
+                        f"{function_call_item.name} "
                         "until bot stops speaking"
                     )
                 else:

api/services/pipecat/realtime/ultravox_realtime.py

@@ -1,19 +1,17 @@
 """Dograh subclass of pipecat's Ultravox realtime LLM service.
 
-Ultravox is audio-native and realtime, but prompt and tool configuration is
-bound to call creation. Dograh therefore cannot lean on in-session updates or
-Gemini-style session resumption handles. This wrapper adapts Ultravox to the
-Dograh engine contract by:
+Ultravox is audio-native and realtime. Its native call stages allow a client
+tool result to atomically change the system prompt and tools while preserving
+the call's server-side conversation history. This wrapper adapts that model to
+the Dograh engine contract by:
 
 - deferring the first call creation until the engine queues the initial node
   opening via ``TTSSpeakFrame`` or ``LLMContextFrame``
-- marking the call for recreation when ``system_instruction`` changes across
-  node transitions, then rebuilding it on the follow-up ``LLMContextFrame``
-  so the transition tool result is present in ``initialMessages``
-- reconstructing Ultravox ``initialMessages`` from Dograh context when the
-  call must be recreated after a node transition
-- appending a transient resumptive user nudge to recreated ``initialMessages``
-  after tool-result transitions, without mutating Dograh's stored context
+- returning node-transition tool results with ``responseType="new-stage"`` so
+  the existing call keeps its complete audio-native history
+- updating the next stage's system prompt and selected tools without a
+  disconnect/reconnect cycle
+- deferring workflow-control tools until any active Ultravox response ends
 - handling Dograh-only frames such as user mute and idle append prompts
 - tagging user transcripts with ``finalized=True`` for downstream parity
 """
@@ -34,12 +32,7 @@ from pipecat.frames.frames import (
     UserMuteStartedFrame,
     UserMuteStoppedFrame,
 )
-from pipecat.processors.aggregators import async_tool_messages
-from pipecat.processors.aggregators.llm_context import (
-    LLMContext,
-    LLMSpecificMessage,
-    is_given,
-)
+from pipecat.processors.aggregators.llm_context import LLMContext, is_given
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import LLMService
 from pipecat.services.settings import _NotGiven, assert_given
@@ -58,10 +51,6 @@ class DograhUltravoxOneShotInputParams(OneShotInputParams):
 
 
 _ULTRAVOX_MAX_TOOL_TIMEOUT_SECS = 40.0
-_RESUMPTION_USER_MESSAGE = (
-    "IMPORTANT: We are resuming an existing conversation. You are given previous turns ONLY for your reference. "
-    "Do not use that to frame your response. Follow your ORIGINAL INSTRUCTIONS ONLY."
-)
 
 
 class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
@@ -72,12 +61,19 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         self._context: LLMContext | None = None
         self._selected_tools = None
         self._user_is_muted: bool = False
-        self._call_system_instruction: str | None = None
-        self._reconnect_required: bool = False
         self._call_started: bool = False
-        self._has_connected_once: bool = False
-        self._pending_reconnect_system_instruction: str | None = None
-        self._pending_initial_messages: list[dict[str, Any]] | None = None
+        self._stage_update_required: bool = False
+        # Ultravox applies a stage update on the matching client tool result,
+        # so retain the provider invocation ID until that result reaches us via
+        # the context aggregator. Unlike Gemini, this ID is part of the wire
+        # protocol needed to update the existing call without reconnecting.
+        self._pending_node_transition_tool_call_ids: set[str] = set()
+        # A stage result can replace the active prompt and tools immediately.
+        # Hold transition invocations separately so ordinary tools can still
+        # run during speech while workflow control waits for response end.
+        self._deferred_node_transition_tool_invocations: list[
+            tuple[str, str, dict[str, Any]]
+        ] = []
         self._pending_user_text_messages: list[str] = []
 
     async def start(self, frame):
@@ -96,9 +92,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         if isinstance(frame, TTSSpeakFrame):
             if not self._socket:
                 await self._connect_call(
-                    system_instruction=self._current_system_instruction(),
                     greeting_text=frame.text,
-                    initial_messages=None,
                     agent_speaks_first=True,
                 )
             else:
@@ -116,18 +110,15 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         changed = await super(UltravoxRealtimeLLMService, self)._update_settings(delta)
         if "output_medium" in changed:
             await self._update_output_medium(assert_given(self._settings.output_medium))
-        if "system_instruction" in changed and self._has_connected_once:
-            # Mirror Gemini's "settings change means reconnect" intent, but
-            # defer the actual new-call creation until the subsequent
-            # LLMContextFrame arrives with the transition tool result. Ultravox
-            # cannot accept that historical tool result over a formal
-            # post-connect tool-response channel the way Gemini can.
-            self._reconnect_required = True
+        if "system_instruction" in changed and self._socket:
+            # The updated instruction is included in the native new-stage
+            # response when the transition tool result reaches _handle_context.
+            self._stage_update_required = True
         handled = {"output_medium", "system_instruction"}
         self._warn_unhandled_updated_settings(changed.keys() - handled)
         return changed
 
-    async def _disconnect(self, preserve_completed_tool_calls: bool = True):
+    async def _disconnect(self):
         self._disconnecting = True
         await self.stop_all_metrics()
         if self._socket:
@@ -136,10 +127,11 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         if self._receive_task:
             await self.cancel_task(self._receive_task, timeout=1.0)
             self._receive_task = None
-        if not preserve_completed_tool_calls:
-            self._completed_tool_calls = set()
+        self._completed_tool_calls = set()
         self._call_started = False
         self._started_placeholder_sent = set()
+        self._pending_node_transition_tool_call_ids = set()
+        self._deferred_node_transition_tool_invocations = []
         self._disconnecting = False
 
     async def _send_user_audio(self, frame):
@@ -149,39 +141,20 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
 
     async def _handle_context(self, context: LLMContext):
         self._context = context
-        system_instruction = self._current_system_instruction()
 
-        if self._socket and not self._reconnect_required:
-            await super()._handle_context(context)
+        if not self._socket:
+            await self._connect_call(
+                greeting_text=None,
+                agent_speaks_first=True,
+            )
             return
 
-        initial_messages, history_tool_call_ids = self._build_initial_messages(context)
-        if history_tool_call_ids:
-            self._completed_tool_calls.update(history_tool_call_ids)
-
-        if self._bot_responding:
-            self._pending_reconnect_system_instruction = system_instruction
-            self._pending_initial_messages = initial_messages
-            return
-
-        await self._reconnect_with_context(
-            system_instruction=system_instruction,
-            initial_messages=initial_messages,
-        )
-
-    async def _handle_response_end(self):
-        await super()._handle_response_end()
-        if self._pending_reconnect_system_instruction is None:
-            return
-
-        system_instruction = self._pending_reconnect_system_instruction
-        initial_messages = self._pending_initial_messages
-        self._pending_reconnect_system_instruction = None
-        self._pending_initial_messages = None
-        await self._reconnect_with_context(
-            system_instruction=system_instruction,
-            initial_messages=initial_messages,
-        )
+        current_tools = self._current_tools_schema(context)
+        if self._pending_node_transition_tool_call_ids and self._tools_changed(
+            current_tools
+        ):
+            self._stage_update_required = True
+        await super()._handle_context(context)
 
     async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
         texts = [
@@ -199,9 +172,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         if not self._socket:
             self._pending_user_text_messages.extend(texts)
             await self._connect_call(
-                system_instruction=self._current_system_instruction(),
                 greeting_text=None,
-                initial_messages=None,
                 agent_speaks_first=False,
             )
             return
@@ -229,17 +200,95 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
             finalized=True,
         )
 
+    def _requires_node_transition_context_aggregation(self) -> bool:
+        """Commit any received final user transcript before changing stages.
+
+        Ultravox preserves its own audio-native history across a stage change,
+        but Dograh's local context still needs the final transcript before the
+        transition handler updates the workflow node.
+        """
+        return True
+
+    async def _handle_tool_invocation(
+        self, tool_name: str, invocation_id: str, parameters: dict[str, Any]
+    ):
+        if self._function_is_node_transition(tool_name):
+            self._pending_node_transition_tool_call_ids.add(invocation_id)
+            if self._bot_responding:
+                self._deferred_node_transition_tool_invocations.append(
+                    (tool_name, invocation_id, parameters)
+                )
+                logger.debug(
+                    f"{self}: deferring workflow-control call {tool_name} "
+                    "until bot turn ends"
+                )
+                return
+        await super()._handle_tool_invocation(tool_name, invocation_id, parameters)
+
+    async def _handle_response_end(self):
+        """Close the current response before applying queued workflow control."""
+        await super()._handle_response_end()
+        await self._run_deferred_node_transition_tool_invocations()
+
+    async def _run_deferred_node_transition_tool_invocations(self):
+        if not self._deferred_node_transition_tool_invocations:
+            return
+
+        invocations = self._deferred_node_transition_tool_invocations
+        self._deferred_node_transition_tool_invocations = []
+        logger.debug(
+            f"{self}: executing {len(invocations)} deferred workflow-control "
+            "call(s) after bot turn ended"
+        )
+        for tool_name, invocation_id, parameters in invocations:
+            await super()._handle_tool_invocation(
+                tool_name, invocation_id, parameters
+            )
+
+    async def _send_tool_result(self, tool_call_id: str, result: str):
+        is_node_transition = tool_call_id in self._pending_node_transition_tool_call_ids
+        try:
+            if is_node_transition and self._stage_update_required:
+                await self._send_node_transition_stage_result(tool_call_id, result)
+            else:
+                await super()._send_tool_result(tool_call_id, result)
+        finally:
+            if is_node_transition:
+                self._pending_node_transition_tool_call_ids.discard(tool_call_id)
+
+    async def _send_node_transition_stage_result(self, tool_call_id: str, result: str):
+        """Apply node settings using Ultravox's native call-stage protocol."""
+        next_tools = self._current_tools_schema(self._context)
+        stage = {
+            "systemPrompt": self._current_system_instruction(),
+            "selectedTools": self._selected_tools_payload(next_tools),
+            # Keep the workflow handler's result as the tool-result message in
+            # the inherited conversation history for the next generation.
+            "toolResultText": result,
+        }
+        logger.debug(
+            f"{self}: updating Ultravox call stage for tool_call_id={tool_call_id} "
+            f"with {len(stage['selectedTools'])} selected tool(s)"
+        )
+        await self._send(
+            {
+                "type": "client_tool_result",
+                "invocationId": tool_call_id,
+                "result": json.dumps(stage, ensure_ascii=True, default=str),
+                "responseType": "new-stage",
+            }
+        )
+        self._selected_tools = next_tools
+        self._stage_update_required = False
+
     async def _connect_call(
         self,
         *,
-        system_instruction: str | None,
         greeting_text: str | None,
-        initial_messages: list[dict[str, Any]] | None,
         agent_speaks_first: bool,
     ):
         params = self._build_one_shot_params(
             greeting_text=greeting_text,
-            initial_messages=initial_messages,
             agent_speaks_first=agent_speaks_first,
         )
         self._params = params
@@ -265,9 +314,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
             logger.info(f"Joining Ultravox Realtime call via URL: {join_url}")
             self._socket = await websocket_client.connect(join_url)
             self._receive_task = self.create_task(self._receive_messages())
-            self._call_system_instruction = system_instruction
             self._call_started = False
-            self._has_connected_once = True
         except Exception as e:
             logger.error(
                 f"{self}: Ultravox call creation/join failed "
@@ -365,40 +412,17 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         for pending_text in pending_texts:
             await self._send_user_text(pending_text)
 
-    async def _reconnect_with_context(
-        self,
-        *,
-        system_instruction: str | None,
-        initial_messages: list[dict[str, Any]] | None,
-    ):
-        call_initial_messages = self._initial_messages_for_call(initial_messages)
-        logger.debug(
-            f"{self}: reconnecting Ultravox call with initialMessages="
-            f"{json.dumps(call_initial_messages, ensure_ascii=True, default=str)}"
-        )
-        if self._socket:
-            await self._disconnect(preserve_completed_tool_calls=True)
-
-        await self._connect_call(
-            system_instruction=system_instruction,
-            greeting_text=None,
-            initial_messages=initial_messages,
-            agent_speaks_first=self._should_agent_speak_first(initial_messages),
-        )
-        self._reconnect_required = False
-
     def _build_one_shot_params(
         self,
         *,
         greeting_text: str | None,
-        initial_messages: list[dict[str, Any]] | None,
         agent_speaks_first: bool,
     ) -> DograhUltravoxOneShotInputParams:
         current_params = self._params
         extra = {
             key: value
             for key, value in current_params.extra.items()
-            if key not in {"firstSpeakerSettings", "initialMessages"}
+            if key != "firstSpeakerSettings"
         }
 
         if greeting_text is not None:
@@ -407,10 +431,6 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
             extra["firstSpeakerSettings"] = {"agent": {}}
         else:
             extra["firstSpeakerSettings"] = {"user": {}}
-        call_initial_messages = self._initial_messages_for_call(initial_messages)
-        if call_initial_messages:
-            extra["initialMessages"] = call_initial_messages
-
         output_medium = self._settings.output_medium
         if isinstance(output_medium, _NotGiven):
             output_medium = current_params.output_medium
@@ -432,6 +452,14 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
             return None
         return context.tools
 
+    def _selected_tools_payload(self, tools: Any) -> list[dict[str, Any]]:
+        return self._to_selected_tools(tools) if tools else []
+
+    def _tools_changed(self, tools: Any) -> bool:
+        return self._selected_tools_payload(tools) != self._selected_tools_payload(
+            self._selected_tools
+        )
+
     def _to_selected_tools(self, tool: Any) -> list[dict[str, Any]]:
         selected_tools = super()._to_selected_tools(tool)
         for selected_tool in selected_tools:
@@ -462,156 +490,6 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         timeout_secs = min(float(item.timeout_secs), _ULTRAVOX_MAX_TOOL_TIMEOUT_SECS)
         return f"{timeout_secs:g}s"
 
-    def _initial_messages_for_call(
-        self, initial_messages: list[dict[str, Any]] | None
-    ) -> list[dict[str, Any]] | None:
-        if not initial_messages:
-            return None
-        if not self._should_add_resumption_user_message(initial_messages):
-            return initial_messages
-
-        return [
-            *initial_messages,
-            {
-                "role": "MESSAGE_ROLE_USER",
-                "text": _RESUMPTION_USER_MESSAGE,
-            },
-        ]
-
-    def _build_initial_messages(
-        self, context: LLMContext
-    ) -> tuple[list[dict[str, Any]] | None, set[str]]:
-        initial_messages: list[dict[str, Any]] = []
-        tool_call_id_to_name: dict[str, str] = {}
-        completed_tool_call_ids: set[str] = set()
-
-        for message in context.get_messages():
-            if isinstance(message, LLMSpecificMessage):
-                continue
-
-            async_payload = async_tool_messages.parse_message(message)
-            if async_payload is not None:
-                if async_payload.kind == "intermediate":
-                    logger.error(
-                        f"{self}: Ultravox does not support streamed async tool results; "
-                        f"dropping intermediate result from initialMessages for "
-                        f"tool_call_id={async_payload.tool_call_id}."
-                    )
-                    continue
-                if async_payload.kind == "final":
-                    initial_message = self._build_ultravox_message(
-                        role="MESSAGE_ROLE_TOOL_RESULT",
-                        text=async_payload.result or "",
-                        invocation_id=async_payload.tool_call_id,
-                        tool_name=tool_call_id_to_name.get(async_payload.tool_call_id),
-                    )
-                    if initial_message is not None:
-                        initial_messages.append(initial_message)
-                    completed_tool_call_ids.add(async_payload.tool_call_id)
-                continue
-
-            role = message.get("role")
-            if role == "user":
-                initial_message = self._build_ultravox_message(
-                    role="MESSAGE_ROLE_USER",
-                    text=self._extract_text_content(message.get("content")),
-                )
-                if initial_message is not None:
-                    initial_messages.append(initial_message)
-            elif role == "assistant":
-                text = self._extract_text_content(message.get("content"))
-                initial_message = self._build_ultravox_message(
-                    role="MESSAGE_ROLE_AGENT",
-                    text=text,
-                )
-                if initial_message is not None:
-                    initial_messages.append(initial_message)
-
-                tool_calls = message.get("tool_calls")
-                if isinstance(tool_calls, list):
-                    for tool_call in tool_calls:
-                        if not isinstance(tool_call, dict):
-                            continue
-                        tool_id = tool_call.get("id")
-                        function = tool_call.get("function")
-                        tool_name = (
-                            function.get("name") if isinstance(function, dict) else None
-                        )
-                        if isinstance(tool_id, str) and isinstance(tool_name, str):
-                            tool_call_id_to_name[tool_id] = tool_name
-                            initial_message = self._build_ultravox_message(
-                                role="MESSAGE_ROLE_TOOL_CALL",
-                                text="",
-                                invocation_id=tool_id,
-                                tool_name=tool_name,
-                            )
-                            if initial_message is not None:
-                                initial_messages.append(initial_message)
-            elif (
-                role == "tool"
-                and message.get("content") != "IN_PROGRESS"
-                and message.get("content") != "CANCELLED"
-            ):
-                tool_call_id = message.get("tool_call_id")
-                initial_message = self._build_ultravox_message(
-                    role="MESSAGE_ROLE_TOOL_RESULT",
-                    text=self._stringify_tool_result(message.get("content")),
-                    invocation_id=tool_call_id
-                    if isinstance(tool_call_id, str)
-                    else None,
-                    tool_name=(
-                        tool_call_id_to_name.get(tool_call_id)
-                        if isinstance(tool_call_id, str)
-                        else None
-                    ),
-                )
-                if initial_message is not None:
-                    initial_messages.append(initial_message)
-                if isinstance(tool_call_id, str):
-                    completed_tool_call_ids.add(tool_call_id)
-
-        return (initial_messages or None), completed_tool_call_ids
-
-    @staticmethod
-    def _build_ultravox_message(
-        *,
-        role: str,
-        text: str | None,
-        invocation_id: str | None = None,
-        tool_name: str | None = None,
-    ) -> dict[str, Any] | None:
-        if text is None:
-            return None
-
-        message: dict[str, Any] = {
-            "role": role,
-            "text": text,
-        }
-        if invocation_id is not None:
-            message["invocationId"] = invocation_id
-        if tool_name is not None:
-            message["toolName"] = tool_name
-        return message
-
-    @staticmethod
-    def _should_agent_speak_first(
-        initial_messages: list[dict[str, Any]] | None,
-    ) -> bool:
-        if not initial_messages:
-            return True
-        return initial_messages[-1].get("role") in {
-            "MESSAGE_ROLE_USER",
-            "MESSAGE_ROLE_TOOL_RESULT",
-        }
-
-    @staticmethod
-    def _should_add_resumption_user_message(
-        initial_messages: list[dict[str, Any]] | None,
-    ) -> bool:
-        if not initial_messages:
-            return False
-        return initial_messages[-1].get("role") == "MESSAGE_ROLE_TOOL_RESULT"
-
     @staticmethod
     def _is_benign_websocket_close(exc: ConnectionClosed) -> bool:
         return any(
@@ -636,18 +514,3 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
                 parts.append(text)
             return "\n".join(parts) if parts else None
         return None
-
-    @staticmethod
-    def _stringify_tool_result(content: Any) -> str:
-        if isinstance(content, str):
-            return content
-        if isinstance(content, list):
-            parts: list[str] = []
-            for part in content:
-                if isinstance(part, dict):
-                    text = part.get("text")
-                    if isinstance(text, str):
-                        parts.append(text)
-            if parts:
-                return "".join(parts)
-        return json.dumps(content, ensure_ascii=True, default=str)