fix(live): stop live runs from writing back into the caller's RunConfig

Two places in the live path wrote into configuration the caller still owns, so
a `RunConfig` came back out of a run holding fields the caller never set, and
a `RunConfig` reused for a later run carried them into it.

The basic request processor aliased two `RunConfig` sub-models straight into
`LiveConnectConfig` rather than copying them, and live request assembly then
mutates both while the session runs: `BaseLlmFlow.run_live` stamps every
server-issued handle onto `session_resumption` when it reconnects and sets
`transparent` there on a Vertex reconnect, and sets
`initial_history_in_client_content` on `history_config` when it seeds a fresh
connection with history. Deep copy both sub-models in `_build_basic_request`.
This matches the treatment `_copy_request_scoped_fields` already gives
`llm_request.config` in the same function, and for the same reason: request
assembly must not write through into configuration the caller still owns. An
absent sub-config stays `None` rather than becoming an empty object.

`Runner.run_live` filled in its AUDIO default by assigning to the caller's
`response_modalities`, so a config that expressed no preference came back out
of the run pinned to AUDIO, and a config reused for a later text run would ask
for audio. Write that default to a copy instead. The copy is shallow: deep
copying a `RunConfig` raises `TypeError: cannot pickle` when `http_options`
holds a live httpx client, and nothing there writes through into a sub-model.

Co-authored-by: Liang Wu <wuliang@google.com>
PiperOrigin-RevId: 966257920

Author: wuliang229

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

@@ -17,8 +17,11 @@
 from __future__ import annotations
 
 from typing import AsyncGenerator
+from typing import Optional
+from typing import TypeVar
 
 from google.genai import types
+from pydantic import BaseModel
 from typing_extensions import override
 
 from ...agents.invocation_context import InvocationContext
@@ -73,6 +76,14 @@ def _copy_http_options(
   )
 
 
+_ModelT = TypeVar('_ModelT', bound=BaseModel)
+
+
+def _copy_or_none(model: Optional[_ModelT]) -> Optional[_ModelT]:
+  """Returns a deep copy of a RunConfig sub-model that assembly then mutates."""
+  return None if model is None else model.model_copy(deep=True)
+
+
 def _copy_request_scoped_fields(
     config: types.GenerateContentConfig,
 ) -> types.GenerateContentConfig:
@@ -183,10 +194,20 @@ def _build_basic_request(
   llm_request.live_connect_config.proactivity = (
       None if is_gemini_3_x else run_config.proactivity
   )
-  llm_request.live_connect_config.session_resumption = (
+  # Copied rather than aliased: live request assembly writes into both of these
+  # while the session runs. `BaseLlmFlow.run_live` stamps each server-issued
+  # resumption handle onto `session_resumption`, and sets
+  # `initial_history_in_client_content` on `history_config` when it seeds a
+  # fresh connection with history. Aliasing the RunConfig's own objects makes
+  # those writes outlive the invocation, so a RunConfig reused for a later run
+  # would carry a stale handle into it. This mirrors what
+  # `_copy_request_scoped_fields` already does for `llm_request.config`.
+  llm_request.live_connect_config.session_resumption = _copy_or_none(
       run_config.session_resumption
   )
-  llm_request.live_connect_config.history_config = run_config.history_config
+  llm_request.live_connect_config.history_config = _copy_or_none(
+      run_config.history_config
+  )
   llm_request.live_connect_config.context_window_compression = (
       run_config.context_window_compression
   )

src/google/adk/runners.py

@@ -1874,7 +1874,15 @@ async def run_live(
     run_config = run_config or RunConfig()
     # Some native audio models requires the modality to be set. So we set it to
     # AUDIO by default.
+    #
+    # The default goes on a copy rather than on the caller's own RunConfig: a
+    # config that asked for nothing in particular would otherwise come back out
+    # of the run pinned to AUDIO, and a config reused for a later run would
+    # carry that choice into it. The copy is shallow on purpose. Deep copying a
+    # RunConfig raises `TypeError: cannot pickle` when `http_options` holds a
+    # live httpx client, and nothing here writes through into a sub-model.
     if run_config.response_modalities is None:
+      run_config = run_config.model_copy()
       run_config.response_modalities = [types.Modality.AUDIO]
     if session is None and (user_id is None or session_id is None):
       raise ValueError(

tests/unittests/flows/llm_flows/test_base_llm_flow.py

@@ -876,6 +876,98 @@ async def mock_receive_2():
       assert invocation_context.live_session_resumption_handle == 'test_handle'
 
 
+@pytest.mark.asyncio
+async def test_reconnect_does_not_write_the_handle_into_the_run_config():
+  """A reconnect must not stamp the server's handle onto the caller's config.
+
+  The reconnect branch assigns the handle onto
+  `llm_request.live_connect_config.session_resumption`. That object comes from
+  the RunConfig, so aliasing it would leave the caller's own RunConfig holding
+  a handle it never set, and reusing that RunConfig for a later run would
+  silently resume this session.
+  """
+
+  real_model = Gemini()
+  mock_connection = mock.AsyncMock()
+
+  async def mock_receive():
+    yield LlmResponse(
+        live_session_resumption_update=types.LiveServerSessionResumptionUpdate(
+            new_handle='server_handle'
+        )
+    )
+    raise ConnectionClosed(None, None)
+
+  mock_connection.receive = mock.Mock(side_effect=mock_receive)
+
+  agent = Agent(name='test_agent', model=real_model)
+  # The caller enables resumption but holds no handle yet, which is how a
+  # first run is configured.
+  run_config_session_resumption = types.SessionResumptionConfig(
+      transparent=True
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent,
+      run_config=RunConfig(session_resumption=run_config_session_resumption),
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = BaseLlmFlowForTesting()
+
+  # `BaseLlmFlow` has no request processors of its own, so the real request
+  # builder has to run for the RunConfig to reach the live connect config at
+  # all. Without it the flow just creates a fresh SessionResumptionConfig and
+  # the aliasing under test never happens.
+  async def mock_preprocess(ctx, req):
+    from google.adk.flows.llm_flows.basic import _build_basic_request
+
+    _build_basic_request(ctx, req)
+    if False:  # pylint: disable=using-constant-test
+      yield
+
+  with (
+      mock.patch.object(flow, '_preprocess_async', side_effect=mock_preprocess),
+      mock.patch.object(flow, '_send_to_model', new_callable=AsyncMock),
+  ):
+    mock_connection_2 = mock.AsyncMock()
+
+    class NonRetryableError(Exception):
+      pass
+
+    async def mock_receive_2():
+      yield LlmResponse(
+          content=types.Content(parts=[types.Part.from_text(text='hi')])
+      )
+      raise NonRetryableError('stop')
+
+    mock_connection_2.receive = mock.Mock(side_effect=mock_receive_2)
+
+    mock_aenter = mock.AsyncMock()
+    mock_aenter.side_effect = [mock_connection, mock_connection_2]
+
+    with mock.patch(
+        'google.adk.models.google_llm.Gemini.connect'
+    ) as mock_connect:
+      mock_connect.return_value.__aenter__ = mock_aenter
+
+      try:
+        async for _ in flow.run_live(invocation_context):
+          pass
+      except NonRetryableError:
+        pass
+
+  # The reconnect happened and carried the handle...
+  assert mock_connect.call_count == 2
+  second_request = mock_connect.call_args_list[1][0][0]
+  assert (
+      second_request.live_connect_config.session_resumption.handle
+      == 'server_handle'
+  )
+  # ...but the caller's own config is untouched.
+  assert run_config_session_resumption.handle is None
+  assert invocation_context.run_config.session_resumption.handle is None
+
+
 @pytest.mark.asyncio
 async def test_run_live_skips_send_history_on_resumption():
   """Test that run_live skips send_history when resuming a session."""

tests/unittests/flows/llm_flows/test_basic_processor.py

@@ -584,3 +584,69 @@ async def test_safety_settings_do_not_accumulate_across_invocations(self):
       pass
 
     assert len(second_request.config.safety_settings) == 1
+
+  @pytest.mark.asyncio
+  async def test_run_config_session_resumption_object_is_not_aliased(self):
+    """The request must not hold the RunConfig's own SessionResumptionConfig.
+
+    `BaseLlmFlow.run_live` stamps every server-issued handle onto the request's
+    `session_resumption`, so aliasing would write those handles back into the
+    caller's RunConfig.
+    """
+    agent = LlmAgent(name='test_agent', model='gemini-1.5-flash')
+    invocation_context = await _create_invocation_context(agent)
+    run_config_session_resumption = types.SessionResumptionConfig(
+        handle='caller_handle'
+    )
+    invocation_context.run_config.session_resumption = (
+        run_config_session_resumption
+    )
+    llm_request = LlmRequest()
+
+    processor = _BasicLlmRequestProcessor()
+    async for _ in processor.run_async(invocation_context, llm_request):
+      pass
+
+    assert (
+        llm_request.live_connect_config.session_resumption.handle
+        == 'caller_handle'
+    )
+    llm_request.live_connect_config.session_resumption.handle = 'server_handle'
+    assert run_config_session_resumption.handle == 'caller_handle'
+
+  @pytest.mark.asyncio
+  async def test_run_config_history_config_object_is_not_aliased(self):
+    """The request must not hold the RunConfig's own HistoryConfig.
+
+    `BaseLlmFlow.run_live` sets `initial_history_in_client_content` on the
+    request when it seeds a fresh connection with history, so aliasing would
+    write that back into the caller's RunConfig.
+    """
+    agent = LlmAgent(name='test_agent', model='gemini-1.5-flash')
+    invocation_context = await _create_invocation_context(agent)
+    run_config_history_config = types.HistoryConfig()
+    invocation_context.run_config.history_config = run_config_history_config
+    llm_request = LlmRequest()
+
+    processor = _BasicLlmRequestProcessor()
+    async for _ in processor.run_async(invocation_context, llm_request):
+      pass
+
+    llm_request.live_connect_config.history_config.initial_history_in_client_content = (
+        True
+    )
+    assert run_config_history_config.initial_history_in_client_content is None
+
+  @pytest.mark.asyncio
+  async def test_absent_live_sub_configs_stay_none(self):
+    """Copying must not turn an unset RunConfig sub-config into an object."""
+    agent = LlmAgent(name='test_agent', model='gemini-1.5-flash')
+    invocation_context = await _create_invocation_context(agent)
+    llm_request = LlmRequest()
+
+    processor = _BasicLlmRequestProcessor()
+    async for _ in processor.run_async(invocation_context, llm_request):
+      pass
+
+    assert llm_request.live_connect_config.session_resumption is None
+    assert llm_request.live_connect_config.history_config is None

tests/unittests/streaming/test_live_streaming_configs.py

@@ -835,3 +835,33 @@ def test_streaming_with_explicit_vad_signal():
   assert (
       llm_request_sent_to_mock.live_connect_config.explicit_vad_signal is True
   )
+
+
+def test_run_live_does_not_write_the_audio_default_into_the_run_config():
+  """The AUDIO default must not be stamped onto the caller's RunConfig.
+
+  `run_live` fills in AUDIO for a caller that expressed no preference. Writing
+  that back into the caller's own RunConfig would hand them a config pinned to
+  AUDIO, so a config reused for a later text run would silently ask for audio.
+  """
+
+  mock_model = testing_utils.MockModel.create([LlmResponse(turn_complete=True)])
+
+  root_agent = Agent(name='root_agent', model=mock_model, tools=[])
+  runner = testing_utils.InMemoryRunner(root_agent=root_agent)
+
+  run_config = RunConfig()
+
+  live_request_queue = LiveRequestQueue()
+  live_request_queue.send_realtime(
+      blob=types.Blob(data=b'\x00\xFF', mime_type='audio/pcm')
+  )
+  runner.run_live(live_request_queue, run_config)
+
+  assert run_config.response_modalities is None
+  assert 'response_modalities' not in run_config.model_fields_set
+  # The run itself still gets the default.
+  assert len(mock_model.requests) == 1
+  assert mock_model.requests[0].live_connect_config.response_modalities == [
+      types.Modality.AUDIO
+  ]