fix: resolve session, memory, GCS, and live issues

src/google/adk/cli/utils/evals.py

@@ -82,7 +82,10 @@ def create_gcs_eval_managers_from_uri(
           ' google-adk[gcp]\nOr: pip install google-cloud-storage>=2.18'
       ) from e
 
-    gcs_bucket = eval_storage_uri.split('://')[1]
+    # Only the bucket name is used; any path segment after the bucket is
+    # ignored, matching the documented "if a path is provided, the bucket
+    # will be extracted" behavior.
+    gcs_bucket = eval_storage_uri.split('://')[1].split('/')[0]
     eval_sets_manager = GcsEvalSetsManager(
         bucket_name=gcs_bucket, project=os.environ['GOOGLE_CLOUD_PROJECT']
     )

src/google/adk/live/__init__.py

@@ -0,0 +1,25 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Live (bidirectional streaming) mode support."""
+
+from __future__ import annotations
+
+from .live_request_queue import LiveRequest as LiveRequest
+from .live_request_queue import LiveRequestQueue as LiveRequestQueue
+
+__all__ = [
+    'LiveRequest',
+    'LiveRequestQueue',
+]

src/google/adk/live/live_request_queue.py

@@ -0,0 +1,97 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+from typing import Optional
+
+from google.genai import types
+from pydantic import BaseModel
+from pydantic import ConfigDict
+
+
+class LiveRequest(BaseModel):
+  """Request send to live agents.
+
+  When multiple fields are set, they are processed by priority (highest first):
+  activity_start > activity_end > audio_stream_end > blob > content.
+  state_delta, if set, is always applied regardless of the other fields.
+  """
+
+  model_config = ConfigDict(ser_json_bytes='base64', val_json_bytes='base64')
+  """The pydantic model config."""
+
+  content: Optional[types.Content] = None
+  """If set, send the content to the model in turn-by-turn mode."""
+
+  blob: Optional[types.Blob] = None
+  """If set, send the blob to the model in realtime mode."""
+
+  activity_start: Optional[types.ActivityStart] = None
+  """If set, signal the start of user activity to the model."""
+
+  activity_end: Optional[types.ActivityEnd] = None
+  """If set, signal the end of user activity to the model."""
+
+  audio_stream_end: bool = False
+  """If set, signal the end of the audio stream to the model. This is only used
+  when Voice Activity Detection is enabled.
+  """
+
+  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."""
+
+  def __init__(self) -> None:
+    self._queue: asyncio.Queue[LiveRequest] = asyncio.Queue()
+
+  def close(self) -> None:
+    self._queue.put_nowait(LiveRequest(close=True))
+
+  def send_content(self, content: types.Content, partial: bool = False) -> None:
+    self._queue.put_nowait(LiveRequest(content=content, partial=partial))
+
+  def send_realtime(self, blob: types.Blob) -> None:
+    self._queue.put_nowait(LiveRequest(blob=blob))
+
+  def send_activity_start(self) -> None:
+    """Sends an activity start signal to mark the beginning of user input."""
+    self._queue.put_nowait(LiveRequest(activity_start=types.ActivityStart()))
+
+  def send_activity_end(self) -> None:
+    """Sends an activity end signal to mark the end of user input."""
+    self._queue.put_nowait(LiveRequest(activity_end=types.ActivityEnd()))
+
+  def send_audio_stream_end(self) -> None:
+    """Sends an audio stream end signal to force flush audio."""
+    self._queue.put_nowait(LiveRequest(audio_stream_end=True))
+
+  def send(self, req: LiveRequest) -> None:
+    self._queue.put_nowait(req)
+
+  async def get(self) -> LiveRequest:
+    return await self._queue.get()

src/google/adk/models/lite_llm.py

@@ -2202,11 +2202,27 @@ def _function_declaration_to_tool_param(
   elif function_declaration.parameters_json_schema:
     parameters = function_declaration.parameters_json_schema
 
+  description = function_declaration.description or ""
+  # Most OpenAI-compatible providers have no dedicated field for a tool's
+  # result/output schema, unlike the Gemini path (see #2828). Rather than
+  # inventing a non-standard key that providers would ignore, surface the
+  # schema by appending it to the description, which is always forwarded.
+  output_schema_dict: Optional[dict[str, Any]] = None
+  if function_declaration.response_json_schema:
+    output_schema_dict = function_declaration.response_json_schema
+  elif function_declaration.response:
+    output_schema_dict = _schema_to_dict(function_declaration.response)
+  if output_schema_dict:
+    description = (
+        f"{description}\n\nResult schema:"
+        f" {json.dumps(output_schema_dict)}"
+    ).strip()
+
   tool_params: dict[str, Any] = {
       "type": "function",
       "function": {
           "name": function_declaration.name,
-          "description": function_declaration.description or "",
+          "description": description,
           "parameters": parameters,
       },
   }
@@ -2490,9 +2506,30 @@ def _message_to_generate_content_response(
     for tool_call in tool_calls:
       if tool_call.type == "function":
         thought_signature = _extract_thought_signature_from_tool_call(tool_call)
+        try:
+          call_args = _parse_tool_call_arguments(tool_call.function.arguments)
+        except json.JSONDecodeError as e:
+          # Malformed/truncated tool-call arguments (e.g. a partial stream
+          # committed to the message) are a recoverable model error here,
+          # not a reason to abort the whole invocation. Log and fall back
+          # to an empty dict so the function-call Part still surfaces and
+          # downstream tool dispatch can retry/recover, instead of the
+          # JSONDecodeError propagating out of response parsing. (The
+          # streaming aggregation path has its own, more specific handling
+          # for arguments truncated by hitting max_output_tokens; this
+          # covers every other malformed-JSON case.)
+          logger.warning(
+              "Failed to parse tool call arguments as JSON for tool"
+              " %r, falling back to an empty dict. Raw arguments: %r."
+              " Error: %s",
+              tool_call.function.name,
+              tool_call.function.arguments,
+              e,
+          )
+          call_args = {}
         part = types.Part.from_function_call(
             name=tool_call.function.name,
-            args=_parse_tool_call_arguments(tool_call.function.arguments),
+            args=call_args,
         )
         function_call = part.function_call
         if function_call is None:

src/google/adk/sessions/_restricted_pickle.py

@@ -65,6 +65,15 @@
     ("datetime", "datetime"),
     ("datetime", "timedelta"),
     ("datetime", "timezone"),
+    # CPython 3.11's `enum.py` pickles some Enum members via
+    # `pickle_by_enum_name`, whose `__reduce_ex__` returns
+    # `(getattr, (self.__class__, self._name_))` instead of the
+    # value-based reconstruction used on 3.10/3.12+. This is safe to allow
+    # unconditionally: `getattr`'s first argument is itself an enum class
+    # resolved through this same `find_class` allow-list check (via its own
+    # GLOBAL/STACK_GLOBAL opcode), so allowing the `getattr` callable here
+    # does not admit any class that wasn't already permitted.
+    ("builtins", "getattr"),
     # Auth models reachable only by subclassing or as a union base, so the
     # annotation walk below does not reach them.
     ("fastapi.openapi.models", "OAuthFlow"),

src/google/adk/sessions/in_memory_session_service.py

@@ -114,7 +114,12 @@ def _create_session_impl(
       state: Optional[dict[str, Any]] = None,
       session_id: Optional[str] = None,
   ) -> Session:
-    if session_id and self._get_session_impl(
+    session_id = (
+        session_id.strip()
+        if session_id and session_id.strip()
+        else platform_uuid.new_uuid()
+    )
+    if self._get_session_impl(
         app_name=app_name, user_id=user_id, session_id=session_id
     ):
       raise AlreadyExistsError(f'Session with id {session_id} already exists.')
@@ -129,11 +134,6 @@ def _create_session_impl(
           user_state_delta
       )
 
-    session_id = (
-        session_id.strip()
-        if session_id and session_id.strip()
-        else platform_uuid.new_uuid()
-    )
     session = Session(
         app_name=app_name,
         user_id=user_id,

src/google/adk/sessions/vertex_ai_session_service.py

@@ -48,7 +48,7 @@
 _COMPACTION_CUSTOM_METADATA_KEY = '_compaction'
 _USAGE_METADATA_CUSTOM_METADATA_KEY = '_usage_metadata'
 
-_SESSION_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$')
+_SESSION_ID_PATTERN = re.compile(r'^[a-z0-9]([a-z0-9-]*[a-z0-9])?$')
 
 
 def _extract_short_session_id(
@@ -568,10 +568,15 @@ def _from_api_event(api_event_obj: vertexai.types.SessionEvent) -> Event:
     event_dict = copy.deepcopy(raw_event_dict)
     timestamp_obj = getattr(api_event_obj, 'timestamp', None)
     event_dict.update({
-        'id': api_event_obj.name.split('/')[-1],
         'invocation_id': getattr(api_event_obj, 'invocation_id', None),
         'author': getattr(api_event_obj, 'author', None),
     })
+    # Preserve the original ADK event id persisted inside raw_event (the
+    # same id streamed to the caller during the live run). Only fall back
+    # to the Vertex resource name for events written before raw_event
+    # carried an id.
+    if not event_dict.get('id'):
+      event_dict['id'] = api_event_obj.name.split('/')[-1]
     if timestamp_obj:
       event_dict['timestamp'] = timestamp_obj.timestamp()
     return Event.model_validate(event_dict)

src/google/adk/tools/preload_memory_tool.py

@@ -52,18 +52,19 @@ async def process_llm_request(
       llm_request: LlmRequest,
   ) -> None:
     user_content = tool_context.user_content
-    if (
-        not user_content
-        or not user_content.parts
-        or not user_content.parts[0].text
-    ):
+    if not user_content or not user_content.parts:
+      return
+
+    user_query = ' '.join(
+        part.text for part in user_content.parts if part.text
+    )
+    if not user_query:
       return
 
-    user_query: str = user_content.parts[0].text
     try:
       response = await tool_context.search_memory(user_query)
     except Exception:
-      logging.warning('Failed to preload memory for query: %s', user_query)
+      logger.warning('Failed to preload memory for query: %s', user_query)
       return
 
     if not response.memories:

src/google/adk/utils/_schema_utils.py

@@ -165,7 +165,10 @@ def validate_node_data(
 
   def _to_serializable(val: Any) -> Any:
     if isinstance(val, BaseModel):
-      return val.model_dump(exclude_none=True)
+      # mode="json" (not the default python mode) so Decimal, datetime,
+      # UUID, and non-str Enum members are converted to JSON-safe values
+      # and any serializer registered with when_used="json" actually runs.
+      return val.model_dump(exclude_none=True, mode="json")
     if isinstance(val, list):
       return [_to_serializable(item) for item in val]
     if isinstance(val, dict):

src/google/adk/workflow/_llm_agent_wrapper.py

@@ -374,12 +374,18 @@ def process_llm_agent_output(
       output = None
   else:
     output = text
+    # Only set when there is no output_schema: this tells the consumer
+    # loop in runners.py that event.content IS the node's output (plain
+    # text), so it can avoid surfacing the same text twice. When
+    # output_schema is set, event.output holds the validated structured
+    # result, which is not the same as the raw message content and must
+    # not be cleared downstream.
+    event.node_info.message_as_output = True
 
   if agent.output_key and output is not None:
     ctx.actions.state_delta[agent.output_key] = output
 
   event.output = output
-  event.node_info.message_as_output = True
 
 
 async def run_llm_agent_as_node(

tests/unittests/sessions/test_vertex_ai_session_service.py

@@ -569,14 +569,14 @@ async def test_get_session_pagination_keeps_client_open():
   session_data = {
       'name': (
           'projects/test-project/locations/test-location/'
-          'reasoningEngines/123/sessions/pagination_test'
+          'reasoningEngines/123/sessions/pagination-test'
       ),
       'update_time': '2024-12-12T12:12:12.123456Z',
       'user_id': 'pagination_user',
   }
-  page1_events = _generate_events_for_page('pagination_test', 0, 100)
-  page2_events = _generate_events_for_page('pagination_test', 100, 100)
-  page3_events = _generate_events_for_page('pagination_test', 200, 50)
+  page1_events = _generate_events_for_page('pagination-test', 0, 100)
+  page2_events = _generate_events_for_page('pagination-test', 100, 100)
+  page3_events = _generate_events_for_page('pagination-test', 200, 50)
 
   mock_client = MockAsyncClientWithPagination(
       session_data=session_data,
@@ -589,7 +589,7 @@ async def test_get_session_pagination_keeps_client_open():
       session_service, '_get_api_client', return_value=mock_client
   ):
     session = await session_service.get_session(
-        app_name='123', user_id='pagination_user', session_id='pagination_test'
+        app_name='123', user_id='pagination_user', session_id='pagination-test'
     )
 
   assert session is not None