fix: fix ultravox node transitions

api/services/pipecat/realtime/ultravox_realtime.py

@@ -11,6 +11,7 @@ the Dograh engine contract by:
   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
 """
@@ -67,6 +68,12 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         # 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):
@@ -124,6 +131,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
         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):
@@ -206,8 +214,37 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
     ):
         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:

api/tests/test_ultravox_realtime_wrapper.py

@@ -204,6 +204,55 @@ async def test_only_registered_node_transition_invocations_are_tracked():
     assert service.run_function_calls.await_count == 2
 
 
+@pytest.mark.asyncio
+async def test_node_transition_invocation_waits_for_response_end():
+    service = _make_service()
+    service.run_function_calls = AsyncMock()
+    service.stop_processing_metrics = AsyncMock()
+    service.push_frame = AsyncMock()
+    service.register_function(
+        "transition_to_next_node",
+        AsyncMock(),
+        is_node_transition=True,
+    )
+    service._bot_responding = "voice"
+
+    await service._handle_tool_invocation(
+        "transition_to_next_node", "call-transition", {"reason": "pricing"}
+    )
+
+    service.run_function_calls.assert_not_awaited()
+    assert service._deferred_node_transition_tool_invocations == [
+        (
+            "transition_to_next_node",
+            "call-transition",
+            {"reason": "pricing"},
+        )
+    ]
+
+    await service._handle_response_end()
+
+    service.run_function_calls.assert_awaited_once()
+    function_call = service.run_function_calls.await_args.args[0][0]
+    assert function_call.function_name == "transition_to_next_node"
+    assert function_call.tool_call_id == "call-transition"
+    assert function_call.arguments == {"reason": "pricing"}
+    assert service._deferred_node_transition_tool_invocations == []
+    assert service._bot_responding is None
+
+
+@pytest.mark.asyncio
+async def test_ordinary_tool_invocation_runs_while_response_is_active():
+    service = _make_service()
+    service.run_function_calls = AsyncMock()
+    service._bot_responding = "voice"
+
+    await service._handle_tool_invocation("lookup_price", "call-lookup", {})
+
+    service.run_function_calls.assert_awaited_once()
+    assert service._deferred_node_transition_tool_invocations == []
+
+
 def test_ultravox_requires_transition_context_aggregation():
     service = _make_service()