feat: add state_delta support to LiveRequest for live mode

The non-live runner path can apply session state changes via a state_delta,
but the live/bidi flow offered no equivalent, so live callers could not seed
or update session state alongside their input. Add an optional state_delta
field to LiveRequest and, in the live flow, apply it as a separate
state-delta event so it always takes effect regardless of content, partial,
or function-response requests. The field defaults to None, preserving
existing behavior.

Close #4220

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 951162326

Author: GWeale

src/google/adk/agents/live_request_queue.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+from typing import Any
 from typing import Optional
 
 from google.genai import types
@@ -32,32 +33,41 @@ class LiveRequest(BaseModel):
   """If set, send the content to the model in turn-by-turn mode.
 
   When multiple fields are set, they are processed by priority (highest first):
-  activity_start > activity_end > blob > content.
+  activity_start > activity_end > blob > content. state_delta, if set, is always
+  applied regardless of the other fields.
   """
   blob: Optional[types.Blob] = None
   """If set, send the blob to the model in realtime mode.
 
   When multiple fields are set, they are processed by priority (highest first):
-  activity_start > activity_end > blob > content.
+  activity_start > activity_end > blob > content. state_delta, if set, is always
+  applied regardless of the other fields.
   """
   activity_start: Optional[types.ActivityStart] = None
   """If set, signal the start of user activity to the model.
 
   When multiple fields are set, they are processed by priority (highest first):
-  activity_start > activity_end > blob > content.
+  activity_start > activity_end > blob > content. state_delta, if set, is always
+  applied regardless of the other fields.
   """
   activity_end: Optional[types.ActivityEnd] = None
   """If set, signal the end of user activity to the model.
 
   When multiple fields are set, they are processed by priority (highest first):
-  activity_start > activity_end > blob > content.
+  activity_start > activity_end > blob > content. state_delta, if set, is always
+  applied regardless of the other fields.
   """
   close: bool = False
   """If set, close the queue. queue.shutdown() is only supported in Python 3.13+."""
 
   partial: bool = False
   """If set, the content is a partial turn update that does not complete the current model turn."""
 
+  state_delta: Optional[dict[str, Any]] = None
+  """If set, these state changes are applied to the session, so they take
+  effect even when the request carries no content or a partial/
+  function-response turn."""
+
 
 class LiveRequestQueue:
   """Queue used to send LiveRequest in a live(bidirectional streaming) way."""

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

@@ -38,6 +38,7 @@
 from ...agents.run_config import StreamingMode
 from ...auth.auth_tool import AuthConfig
 from ...events.event import Event
+from ...events.event_actions import EventActions
 from ...models.base_llm_connection import BaseLlmConnection
 from ...models.google_llm import Gemini
 from ...models.google_llm import GoogleLLMVariant
@@ -795,6 +796,29 @@ async def _send_to_model(
       # Yield to event loop for cooperative multitasking
       await asyncio.sleep(0)
 
+      # State changes ride on the user content event when one is created below;
+      # otherwise a standalone content-less event applies them.
+      is_function_response = bool(
+          live_request.content
+          and live_request.content.parts
+          and any(part.function_response for part in live_request.content.parts)
+      )
+      content_event_created = bool(
+          live_request.content
+          and not live_request.close
+          and not live_request.partial
+          and not is_function_response
+      )
+      if live_request.state_delta and not content_event_created:
+        await invocation_context.session_service.append_event(
+            session=invocation_context.session,
+            event=Event(
+                invocation_id=invocation_context.invocation_id,
+                author='user',
+                actions=EventActions(state_delta=live_request.state_delta),
+            ),
+        )
+
       if live_request.close:
         await llm_connection.close()
         return
@@ -817,9 +841,6 @@ async def _send_to_model(
           raise ValueError('User message cannot contain function calls.')
         # Persist user text content to session (similar to non-live mode)
         # Skip function responses - they are already handled separately
-        is_function_response = content.parts and any(
-            part.function_response for part in content.parts
-        )
         if not is_function_response and not content.role:
           content.role = 'user'
         if not is_function_response and not live_request.partial:
@@ -828,6 +849,9 @@ async def _send_to_model(
               invocation_id=invocation_context.invocation_id,
               author='user',
               content=content,
+              actions=EventActions(state_delta=live_request.state_delta)
+              if live_request.state_delta
+              else EventActions(),
           )
           await invocation_context.session_service.append_event(
               session=invocation_context.session,

tests/unittests/agents/test_live_request_queue.py

@@ -79,3 +79,13 @@ async def test_get():
 
     assert result == res
     mock_get.assert_called_once()
+
+
+def test_state_delta_defaults_to_none():
+  assert LiveRequest().state_delta is None
+
+
+def test_state_delta_json_round_trip():
+  req = LiveRequest(state_delta={"a": 1})
+  restored = LiveRequest.model_validate_json(req.model_dump_json())
+  assert restored.state_delta == {"a": 1}

tests/unittests/flows/llm_flows/test_base_llm_flow_realtime.py

@@ -230,3 +230,143 @@ async def test_send_to_model_with_intermediate_text_content(
       content, partial=True
   )
   invocation_context.session_service.append_event.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_send_to_model_applies_state_delta(mock_llm_connection):
+  """Test _send_to_model applies state_delta as a state-delta event."""
+  agent = Agent(name='test_agent', model='mock')
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = TestBaseLlmFlow()
+
+  invocation_context.live_request_queue.send(
+      LiveRequest(state_delta={'k': 'v'})
+  )
+  invocation_context.live_request_queue.close()
+
+  await flow._send_to_model(mock_llm_connection, invocation_context)
+
+  assert invocation_context.session.state['k'] == 'v'
+  mock_llm_connection._send_content.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_send_to_model_state_delta_with_content(mock_llm_connection):
+  """Test _send_to_model applies state_delta and forwards content together."""
+  agent = Agent(name='test_agent', model='mock')
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = TestBaseLlmFlow()
+
+  content = types.Content(role='user', parts=[types.Part.from_text(text='hi')])
+  invocation_context.live_request_queue.send(
+      LiveRequest(content=content, state_delta={'k': 'v'})
+  )
+  invocation_context.live_request_queue.close()
+
+  await flow._send_to_model(mock_llm_connection, invocation_context)
+
+  assert invocation_context.session.state['k'] == 'v'
+  # The state delta rides on the single user content event (no extra event).
+  events = invocation_context.session.events
+  assert len(events) == 1
+  assert events[0].content == content
+  assert events[0].actions.state_delta == {'k': 'v'}
+  mock_llm_connection._send_content.assert_called_once_with(
+      content, partial=False
+  )
+
+
+@pytest.mark.asyncio
+async def test_send_to_model_state_delta_with_partial_content(
+    mock_llm_connection,
+):
+  """state_delta applies even when the partial turn skips the content event."""
+  agent = Agent(name='test_agent', model='mock')
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = TestBaseLlmFlow()
+
+  content = types.Content(
+      role='user', parts=[types.Part.from_text(text='progress')]
+  )
+  invocation_context.live_request_queue.send(
+      LiveRequest(content=content, state_delta={'k': 'v'}, partial=True)
+  )
+  invocation_context.live_request_queue.close()
+
+  await flow._send_to_model(mock_llm_connection, invocation_context)
+
+  assert invocation_context.session.state['k'] == 'v'
+  # The partial content does not create a user content event.
+  assert all(e.content is None for e in invocation_context.session.events)
+  mock_llm_connection._send_content.assert_called_once_with(
+      content, partial=True
+  )
+
+
+@pytest.mark.asyncio
+async def test_send_to_model_state_delta_with_function_response(
+    mock_llm_connection,
+):
+  """state_delta applies even when the content is a function response."""
+  agent = Agent(name='test_agent', model='mock')
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = TestBaseLlmFlow()
+
+  content = types.Content(
+      role='user',
+      parts=[
+          types.Part.from_function_response(
+              name='tool', response={'result': 'ok'}
+          )
+      ],
+  )
+  invocation_context.live_request_queue.send(
+      LiveRequest(content=content, state_delta={'k': 'v'})
+  )
+  invocation_context.live_request_queue.close()
+
+  await flow._send_to_model(mock_llm_connection, invocation_context)
+
+  assert invocation_context.session.state['k'] == 'v'
+  # Function responses do not create a user content event.
+  assert all(e.content is None for e in invocation_context.session.events)
+  mock_llm_connection._send_content.assert_called_once_with(
+      content, partial=False
+  )
+
+
+@pytest.mark.asyncio
+async def test_send_to_model_state_delta_with_close(mock_llm_connection):
+  """state_delta is flushed even when the request also closes the connection."""
+  agent = Agent(name='test_agent', model='mock')
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, user_content=''
+  )
+  invocation_context.live_request_queue = LiveRequestQueue()
+
+  flow = TestBaseLlmFlow()
+
+  invocation_context.live_request_queue.send(
+      LiveRequest(state_delta={'k': 'v'}, close=True)
+  )
+
+  await flow._send_to_model(mock_llm_connection, invocation_context)
+
+  assert invocation_context.session.state['k'] == 'v'
+  mock_llm_connection.close.assert_called_once()