fix: yield tool_call_parts immediately in live mode to unblock Gemini 3.1 tool calls

Fixes #5407

END_PUBLIC

Merge #5408

## Summary

- Tool call parts received via `LiveServerToolCall` in `GeminiLlmConnection.receive()` are now yielded immediately instead of being deferred until `turn_complete`
- This unblocks function/tool calling for Gemini 3.1 Flash Live models which do not emit `turn_complete` until they receive the tool response

## Problem

Gemini 3.1 Flash Live models send tool calls via `LiveServerToolCall` and wait for the tool response before sending `turn_complete`. The current code accumulates `tool_call_parts` and only yields them on `turn_complete` or loop exit, creating a deadlock where tools are never executed.

## Fix

Yield `tool_call_parts` immediately after receiving `message.tool_call`. This is backward-compatible — earlier models that send `turn_complete` after tool calls will still work because the existing yield paths become no-ops when `tool_call_parts` is already empty.

## Test plan

- [x] Tested with `gemini-3.1-flash-live-preview` in live mode — tool calls now execute on the first message and the model responds with audio after receiving tool results
- [x] Verified no regression with the existing `turn_complete`-based flow

Co-authored-by: Sasha Sobran <asobran@google.com>
COPYBARA_INTEGRATE_REVIEW=#5408 from sandeshveerani4:fix/live-tool-call-yield ca0e760
PiperOrigin-RevId: 908326192

Author: 2 people

src/google/adk/models/gemini_llm_connection.py

@@ -366,6 +366,25 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
               types.Part(function_call=function_call)
               for function_call in message.tool_call.function_calls
           ])
+          # Gemini 3.1 does not emit turn_complete until it receives the
+          # tool response, so yield tool calls immediately to avoid
+          # deadlocking the conversation. Other models (e.g. 2.5-pro,
+          # native-audio) send turn_complete after tool calls, so buffer
+          # and merge them into a single response at turn_complete.
+          if (
+              model_name_utils.is_gemini_3_1_flash_live(self._model_version)
+              and tool_call_parts
+          ):
+            logger.debug(
+                'Yielding tool_call_parts immediately for Gemini 3.1 live tool'
+                ' call'
+            )
+            yield LlmResponse(
+                content=types.Content(role='model', parts=tool_call_parts),
+                model_version=self._model_version,
+                live_session_id=live_session_id,
+            )
+            tool_call_parts = []
         if message.session_resumption_update:
           logger.debug('Received session resumption message: %s', message)
           yield (

tests/unittests/models/test_gemini_llm_connection.py

@@ -1136,6 +1136,49 @@ async def mock_receive_generator():
   assert responses[1].turn_complete is True
 
 
+@pytest.mark.asyncio
+async def test_receive_tool_calls_yielded_immediately_for_gemini_3_1(
+    mock_gemini_session,
+):
+  """Test that tool calls are yielded immediately for Gemini 3.1."""
+  connection = GeminiLlmConnection(
+      mock_gemini_session,
+      api_backend=GoogleLLMVariant.VERTEX_AI,
+      model_version='gemini-3.1-flash-live-preview',
+  )
+
+  mock_tool_call_msg = mock.create_autospec(
+      types.LiveServerMessage, instance=True
+  )
+  mock_tool_call_msg.usage_metadata = None
+  mock_tool_call_msg.server_content = None
+  mock_tool_call_msg.session_resumption_update = None
+  mock_tool_call_msg.go_away = None
+
+  function_call = types.FunctionCall(
+      name='test_tool',
+      args={'arg': 'value'},
+  )
+  mock_tool_call = mock.create_autospec(types.LiveServerToolCall, instance=True)
+  mock_tool_call.function_calls = [function_call]
+  mock_tool_call_msg.tool_call = mock_tool_call
+
+  async def mock_receive_generator():
+    yield mock_tool_call_msg
+
+  receive_mock = mock.Mock(return_value=mock_receive_generator())
+  mock_gemini_session.receive = receive_mock
+
+  responses = []
+  async for resp in connection.receive():
+    responses.append(resp)
+    break
+
+  assert len(responses) == 1
+  assert responses[0].content is not None
+  assert responses[0].content.parts[0].function_call.name == 'test_tool'
+
+
 @pytest.mark.asyncio
 async def test_receive_go_away(gemini_connection, mock_gemini_session):
   """Test receive yields go_away message."""