fix(live): Resolve 1007 error and support Gemini 3.1 Flash Live protocol

Resolves protocol handling discrepancies when connecting to Gemini 3.1 Flash Live models:
- Sets initial_history_in_client_content=True when seeding conversation history during connection handshake.
- Appends turn_complete=True on the final history turn during setup.
- Iterates sequentially through all parts of model_turn to unpack multiplexed audio and text responses.
- Prunes unsupported proactivity and affective dialogue configurations when assembling LiveConnectConfig.

Change-Id: I2d0ff38d8a6eb40ea17b37f65a4ddd093230842c

Author: sasha-gitg

src/google/adk/flows/llm_flows/base_llm_flow.py

@@ -51,6 +51,7 @@
 from ...tools.base_toolset import BaseToolset
 from ...tools.tool_context import ToolContext
 from ...utils.context_utils import Aclosing
+from ...utils import model_name_utils
 from .audio_cache_manager import AudioCacheManager
 from .functions import build_auth_request_event
 
@@ -552,6 +553,20 @@ async def run_live(
             if session_resumption.transparent is None:
               session_resumption.transparent = True
 
+        if (
+            isinstance(llm, Gemini)
+            and llm._api_backend == GoogleLLMVariant.GEMINI_API
+            and model_name_utils.is_gemini_3_1_flash_live(llm_request.model)
+            and llm_request.contents
+            and not invocation_context.live_session_resumption_handle
+        ):
+          if llm_request.live_connect_config is None:
+            llm_request.live_connect_config = types.LiveConnectConfig()
+          if llm_request.live_connect_config.history_config is None:
+            llm_request.live_connect_config.history_config = types.HistoryConfig(
+                initial_history_in_client_content=True
+            )
+
         logger.info(
             'Establishing live connection for agent: %s',
             invocation_context.agent.name,

src/google/adk/flows/llm_flows/basic.py

@@ -25,6 +25,7 @@
 from ...agents.invocation_context import InvocationContext
 from ...events.event import Event
 from ...models.llm_request import LlmRequest
+from ...utils import model_name_utils
 from ...utils.output_schema_utils import can_use_output_schema_with_tools
 from ._base_llm_processor import BaseLlmRequestProcessor
 
@@ -82,11 +83,13 @@ def _build_basic_request(
   llm_request.live_connect_config.realtime_input_config = (
       invocation_context.run_config.realtime_input_config
   )
+  active_model_name = getattr(getattr(agent, 'canonical_live_model', None), 'model', None) or llm_request.model
+  is_gemini_31 = model_name_utils.is_gemini_3_1_flash_live(active_model_name)
   llm_request.live_connect_config.enable_affective_dialog = (
-      invocation_context.run_config.enable_affective_dialog
+      None if is_gemini_31 else invocation_context.run_config.enable_affective_dialog
   )
   llm_request.live_connect_config.proactivity = (
-      invocation_context.run_config.proactivity
+      None if is_gemini_31 else invocation_context.run_config.proactivity
   )
   llm_request.live_connect_config.session_resumption = (
       invocation_context.run_config.session_resumption

src/google/adk/models/gemini_llm_connection.py

@@ -80,10 +80,24 @@ async def send_history(self, history: list[types.Content]):
     ]
 
     if contents:
+      is_gemini_31 = model_name_utils.is_gemini_3_1_flash_live(
+          self._model_version
+      )
+      # Gemini Enterprise Agent Platform does not support history_config in the SDK.
+      # To initialize a live session with prior history without hitting a 1007
+      # protocol error (invalid role mid-session), we consolidate previous multi-turn
+      # interactions into a unified contextual preamble on a single user role turn.
+      if is_gemini_31 and self._api_backend != GoogleLLMVariant.GEMINI_API:
+        collapsed_text = "Previous conversation history:\n"
+        for c in contents:
+          text_parts = "".join(p.text for p in c.parts if p.text)
+          collapsed_text += f'[{c.role}]: {text_parts}\n'
+        contents = [types.Content(role='user', parts=[types.Part.from_text(text=collapsed_text)])]
+
       logger.debug('Sending history to live connection: %s', contents)
       await self._gemini_session.send_client_content(
           turns=contents,
-          turn_complete=contents[-1].role == 'user',
+          turn_complete=True if is_gemini_31 else (contents[-1].role == 'user'),
       )
     else:
       logger.info('no content is sent')
@@ -254,18 +268,20 @@ async def receive(self) -> AsyncGenerator[LlmResponse, None]:
               llm_response.grounding_metadata = (
                   message.server_content.grounding_metadata
               )
-            if content.parts[0].text:
-              current_is_thought = getattr(content.parts[0], 'thought', False)
-              if text and current_is_thought != is_thought:
-                yield self.__build_full_text_response(text, is_thought)
-                text = ''
-                is_thought = False
-
-              text += content.parts[0].text
-              is_thought = current_is_thought
-              llm_response.partial = True
+            has_inline_data = any(p.inline_data for p in content.parts)
+            for part in content.parts:
+              if part.text:
+                current_is_thought = getattr(part, 'thought', False)
+                if text and current_is_thought != is_thought:
+                  yield self.__build_full_text_response(text, is_thought)
+                  text = ''
+                  is_thought = False
+
+                text += part.text
+                is_thought = current_is_thought
+                llm_response.partial = True
             # don't yield the merged text event when receiving audio data
-            elif text and not content.parts[0].inline_data:
+            if text and not any(p.text for p in content.parts) and not has_inline_data:
               yield self.__build_full_text_response(text, is_thought)
               text = ''
               is_thought = False

tests/unittests/flows/llm_flows/test_base_llm_flow.py

@@ -24,7 +24,7 @@
 from google.adk.events.event import Event
 from google.adk.flows.llm_flows.base_llm_flow import _handle_after_model_callback
 from google.adk.flows.llm_flows.base_llm_flow import BaseLlmFlow
-from google.adk.models.google_llm import Gemini
+from google.adk.models.google_llm import Gemini, GoogleLLMVariant
 from google.adk.models.llm_request import LlmRequest
 from google.adk.models.llm_response import LlmResponse
 from google.adk.plugins.base_plugin import BasePlugin
@@ -1386,3 +1386,75 @@ async def mock_receive_2():
         second_call_req = mock_connect.call_args_list[1][0][0]
         session_resump = second_call_req.live_connect_config.session_resumption
         assert session_resump.transparent
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "api_backend,should_have_history_config",
+    [
+        (GoogleLLMVariant.GEMINI_API, True),
+        (GoogleLLMVariant.VERTEX_AI, False),
+    ],
+)
+async def test_run_live_history_config_gated_by_backend(
+    api_backend, should_have_history_config
+):
+  """Test that run_live only sets history_config for Gemini API backend."""
+
+  real_model = Gemini(model='gemini-3.1-flash-live-preview')
+  mock_connection = mock.AsyncMock()
+
+  class StopTestError(Exception):
+    pass
+
+  async def mock_receive():
+    yield LlmResponse(
+        content=types.Content(parts=[types.Part.from_text(text='hi')])
+    )
+    raise StopTestError('stop')
+
+  mock_connection.receive = mock.Mock(side_effect=mock_receive)
+
+  agent = Agent(name='test_agent', model=real_model)
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = BaseLlmFlowForTesting()
+
+  with mock.patch.object(flow, '_send_to_model', new_callable=AsyncMock):
+    async def mock_preprocess(ctx, req):
+      req.contents = [types.Content(parts=[types.Part.from_text(text='history')])]
+      yield Event(id=Event.new_id(), author='test')
+
+    with mock.patch.object(
+        flow, '_preprocess_async', side_effect=mock_preprocess
+    ):
+      with mock.patch.object(
+          Gemini, '_api_backend', new_callable=mock.PropertyMock
+      ) as mock_backend:
+        mock_backend.return_value = api_backend
+        with mock.patch(
+            'google.adk.models.google_llm.Gemini.connect'
+        ) as mock_connect:
+          mock_connect.return_value.__aenter__.return_value = mock_connection
+
+          try:
+            async for _ in flow.run_live(invocation_context):
+              pass
+          except StopTestError:
+            pass
+
+          assert mock_connect.call_count == 1
+          called_req = mock_connect.call_args[0][0]
+          if should_have_history_config:
+            assert called_req.live_connect_config is not None
+            assert called_req.live_connect_config.history_config is not None
+            assert (
+                called_req.live_connect_config.history_config.initial_history_in_client_content
+                is True
+            )
+          else:
+            if called_req.live_connect_config:
+              assert called_req.live_connect_config.history_config is None

tests/unittests/models/test_gemini_llm_connection.py

@@ -1540,3 +1540,95 @@ async def mock_receive_generator():
       responses[0].turn_complete_reason
       == types.TurnCompleteReason.RESPONSE_REJECTED
   )
+
+
+@pytest.mark.asyncio
+async def test_receive_multiplexed_parts(gemini_connection, mock_gemini_session):
+  """Test receive with multiplexed inline data and text content."""
+  mock_content = types.Content(
+      role='model',
+      parts=[
+          types.Part(
+              inline_data=types.Blob(data=b'audio_data', mime_type='audio/pcm')
+          ),
+          types.Part.from_text(text='transcription text'),
+      ],
+  )
+  mock_server_content = mock.Mock()
+  mock_server_content.model_turn = mock_content
+  mock_server_content.interrupted = False
+  mock_server_content.input_transcription = None
+  mock_server_content.output_transcription = None
+  mock_server_content.turn_complete = False
+  mock_server_content.grounding_metadata = None
+
+  mock_message = mock.AsyncMock()
+  mock_message.usage_metadata = None
+  mock_message.server_content = mock_server_content
+  mock_message.tool_call = None
+  mock_message.session_resumption_update = None
+  mock_message.go_away = None
+
+  async def mock_receive_generator():
+    yield mock_message
+
+  receive_mock = mock.Mock(return_value=mock_receive_generator())
+  mock_gemini_session.receive = receive_mock
+
+  responses = [resp async for resp in gemini_connection.receive()]
+
+  assert responses
+  content_response = next((r for r in responses if r.content), None)
+  assert content_response is not None
+  assert content_response.content == mock_content
+  assert content_response.partial is True
+
+
+@pytest.mark.asyncio
+async def test_send_history_gemini_31_turn_complete(mock_gemini_session):
+  """Verify Gemini 3.1 Live history seeding explicitly appends turn_complete=True."""
+  from google.adk.models.google_llm import GoogleLLMVariant
+  conn = GeminiLlmConnection(
+      mock_gemini_session,
+      api_backend=GoogleLLMVariant.GEMINI_API,
+      model_version='gemini-3.1-flash-live-preview',
+  )
+  mock_gemini_session.send_client_content = mock.AsyncMock()
+
+  mock_contents = [
+      types.Content(role='user', parts=[types.Part.from_text(text='hi')]),
+      types.Content(role='model', parts=[types.Part.from_text(text='hello')]),
+  ]
+  await conn.send_history(mock_contents)
+
+  mock_gemini_session.send_client_content.assert_called_once_with(
+      turns=mock_contents,
+      turn_complete=True,
+  )
+
+
+@pytest.mark.asyncio
+async def test_send_history_collapse_vertex_ai(mock_gemini_session):
+  """Verify history prompt collapse when seeding Gemini 3.1 Live on Vertex AI backend."""
+  from google.adk.models.google_llm import GoogleLLMVariant
+  conn = GeminiLlmConnection(
+      mock_gemini_session,
+      api_backend=GoogleLLMVariant.VERTEX_AI,
+      model_version='gemini-3.1-flash-live-preview',
+  )
+  mock_gemini_session.send_client_content = mock.AsyncMock()
+
+  mock_contents = [
+      types.Content(role='user', parts=[types.Part.from_text(text='hi')]),
+      types.Content(role='model', parts=[types.Part.from_text(text='hello')]),
+  ]
+  await conn.send_history(mock_contents)
+
+  assert mock_gemini_session.send_client_content.call_count == 1
+  called_turns = mock_gemini_session.send_client_content.call_args.kwargs['turns']
+  assert len(called_turns) == 1
+  assert called_turns[0].role == 'user'
+  assert 'Previous conversation history:' in called_turns[0].parts[0].text
+  assert '[user]: hi' in called_turns[0].parts[0].text
+  assert '[model]: hello' in called_turns[0].parts[0].text
+  assert mock_gemini_session.send_client_content.call_args.kwargs['turn_complete'] is True