fix: Port session id validation, thread id scoping and Vertex RAG cleanup fixes to v1

src/google/adk/agents/langgraph_agent.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import hashlib
 from typing import AsyncGenerator
 from typing import Union
 
@@ -31,6 +32,33 @@
 from .invocation_context import InvocationContext
 
 
+def _get_thread_id(app_name: str, user_id: str, session_id: str) -> str:
+  """Derives the LangGraph checkpointer thread id for a session.
+
+  Session ids are caller-chosen and are only unique within an
+  (app_name, user_id) pair, so all three components have to take part in the
+  thread id. Each component is length-prefixed before hashing so that a
+  component containing the separator cannot stand in for a different triple.
+  The composite is hashed rather than used verbatim so that the thread id is a
+  fixed-length token no checkpointer backend has to escape, and so the user id
+  is not written into checkpointer storage; the cost is that a stored row can
+  only be tied back to a session by recomputing the digest.
+
+  Args:
+    app_name: the app the session belongs to
+    user_id: the user the session belongs to
+    session_id: the session id
+
+  Returns:
+    a deterministic thread id for the session
+  """
+  key = '|'.join(
+      f'{len(component)}:{component}'
+      for component in (app_name, user_id, session_id)
+  )
+  return hashlib.sha256(key.encode('utf-8')).hexdigest()
+
+
 def _get_last_human_messages(events: list[Event]) -> list[HumanMessage]:
   """Extracts last human messages from given list of events.
 
@@ -50,7 +78,14 @@ def _get_last_human_messages(events: list[Event]) -> list[HumanMessage]:
 
 
 class LangGraphAgent(BaseAgent):
-  """Currently a concept implementation, supports single and multi-turn."""
+  """Currently a concept implementation, supports single and multi-turn.
+
+  The checkpointer thread id is derived from the session's app name, user id
+  and id together, because session ids are only unique within an
+  (app name, user id) pair. Checkpoints written by earlier releases, which
+  keyed the thread on the session id alone, are not reused: with a persistent
+  checkpointer the first turn after upgrading resumes from empty graph state.
+  """
 
   model_config = ConfigDict(
       arbitrary_types_allowed=True,
@@ -68,7 +103,13 @@ async def _run_async_impl(
   ) -> AsyncGenerator[Event, None]:
 
     # Needed for langgraph checkpointer (for subsequent invocations; multi-turn)
-    config: RunnableConfig = {'configurable': {'thread_id': ctx.session.id}}
+    config: RunnableConfig = {
+        'configurable': {
+            'thread_id': _get_thread_id(
+                ctx.session.app_name, ctx.session.user_id, ctx.session.id
+            )
+        }
+    }
 
     # Add instruction as SystemMessage if graph state is empty
     current_graph_state = self.graph.get_state(config)

src/google/adk/memory/in_memory_memory_service.py

@@ -33,8 +33,8 @@
 _UNKNOWN_SESSION_ID = '__unknown_session_id__'
 
 
-def _user_key(app_name: str, user_id: str) -> str:
-  return f'{app_name}/{user_id}'
+def _user_key(app_name: str, user_id: str) -> tuple[str, str]:
+  return (app_name, user_id)
 
 
 def _extract_words_lower(text: str) -> set[str]:
@@ -54,8 +54,8 @@ class InMemoryMemoryService(BaseMemoryService):
   def __init__(self):
     self._lock = threading.Lock()
 
-    self._session_events: dict[str, dict[str, list[Event]]] = {}
-    """Keys are "{app_name}/{user_id}". Values are dicts of session_id to
+    self._session_events: dict[tuple[str, str], dict[str, list[Event]]] = {}
+    """Keys are (app_name, user_id). Values are dicts of session_id to
     session event lists.
     """
 

src/google/adk/memory/vertex_ai_rag_memory_service.py

@@ -15,6 +15,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import binascii
 from collections import OrderedDict
@@ -118,48 +119,67 @@ def __init__(
 
   @override
   async def add_session_to_memory(self, session: Session) -> None:
-    with tempfile.NamedTemporaryFile(
-        mode="w", delete=False, suffix=".txt"
-    ) as temp_file:
-
-      output_lines = []
-      for event in session.events:
-        if not event.content or not event.content.parts:
-          continue
-        text_parts = [
-            part.text.replace("\n", " ")
-            for part in event.content.parts
-            if part.text
-        ]
-        if text_parts:
-          output_lines.append(
-              json.dumps({
-                  "author": event.author,
-                  "timestamp": event.timestamp,
-                  "text": ".".join(text_parts),
-              })
-          )
-      output_string = "\n".join(output_lines)
-      temp_file.write(output_string)
-      temp_file_path = temp_file.name
-
-    if not self._vertex_rag_store.rag_resources:
-      raise ValueError("Rag resources must be set.")
+    rag_resources = self._vertex_rag_store.rag_resources or ()
+    corpus_names = tuple(
+        resource.rag_corpus for resource in rag_resources if resource.rag_corpus
+    )
+    if not corpus_names or len(corpus_names) != len(rag_resources):
+      raise ValueError("rag_corpus must be set on every RAG resource.")
 
-    from ..dependencies.vertexai import rag
+    output_lines = []
+    for event in session.events:
+      if not event.content or not event.content.parts:
+        continue
+      text_parts = [
+          part.text.replace("\n", " ")
+          for part in event.content.parts
+          if part.text
+      ]
+      if text_parts:
+        output_lines.append(
+            json.dumps({
+                "author": event.author,
+                "timestamp": event.timestamp,
+                "text": ".".join(text_parts),
+            })
+        )
+    output_string = "\n".join(output_lines)
 
-    for rag_resource in self._vertex_rag_store.rag_resources:
-      rag.upload_file(
-          corpus_name=rag_resource.rag_corpus,
-          path=temp_file_path,
-          # this is the temp workaround as upload file does not support
-          # adding metadata, thus use display_name to store the session info.
-          display_name=_build_source_display_name(
-              session.app_name, session.user_id, session.id
-          ),
-      )
+    from ..dependencies.vertexai import rag
 
-    os.remove(temp_file_path)
+    temp_file_path: str | None = None
+    try:
+      with tempfile.NamedTemporaryFile(
+          mode="w",
+          delete=False,
+          encoding="utf-8",
+          suffix=".txt",
+      ) as temp_file:
+        temp_file_path = temp_file.name
+        temp_file.write(output_string)
+
+      # Fails fast: the RAG API cannot roll back corpora already written.
+      for corpus_name in corpus_names:
+        # The SDK call is synchronous, so it runs on a worker thread rather
+        # than blocking the event loop for the whole HTTP round trip.
+        await asyncio.to_thread(
+            rag.upload_file,
+            corpus_name=corpus_name,
+            path=temp_file_path,
+            # this is the temp workaround as upload file does not support
+            # adding metadata, thus use display_name to store the session info.
+            display_name=_build_source_display_name(
+                session.app_name, session.user_id, session.id
+            ),
+        )
+    finally:
+      # The transcript is plaintext, so it is removed on failure and
+      # cancellation as well as on success.
+      if temp_file_path:
+        try:
+          os.remove(temp_file_path)
+        except FileNotFoundError:
+          pass
 
   @override
   async def search_memory(
@@ -169,7 +189,10 @@ async def search_memory(
     from ..dependencies.vertexai import rag
     from ..events.event import Event
 
-    response = rag.retrieval_query(
+    # The SDK call is synchronous, so it runs on a worker thread rather than
+    # blocking the event loop for the whole HTTP round trip.
+    response = await asyncio.to_thread(
+        rag.retrieval_query,
         text=query,
         rag_resources=self._vertex_rag_store.rag_resources,
         rag_corpora=self._vertex_rag_store.rag_corpora,

src/google/adk/sessions/vertex_ai_session_service.py

@@ -47,6 +47,19 @@
 _COMPACTION_CUSTOM_METADATA_KEY = '_compaction'
 _USAGE_METADATA_CUSTOM_METADATA_KEY = '_usage_metadata'
 
+_SESSION_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]+$')
+
+
+def _validate_session_id(session_id: str) -> None:
+  """Rejects session IDs that could escape the URL path segment."""
+  if not isinstance(session_id, str) or not _SESSION_ID_PATTERN.fullmatch(
+      session_id
+  ):
+    raise ValueError(
+        f'Invalid session_id {session_id!r}: must match'
+        f' {_SESSION_ID_PATTERN.pattern}.'
+    )
+
 
 def _quote_filter_literal(value: str) -> str:
   """Quotes filter values so embedded metacharacters stay inside the literal."""
@@ -127,6 +140,7 @@ async def create_session(
 
     config = {'session_state': state} if state else {}
     if session_id:
+      _validate_session_id(session_id)
       config['session_id'] = session_id
     config.update(kwargs)
     async with self._get_api_client() as api_client:
@@ -157,6 +171,7 @@ async def get_session(
       session_id: str,
       config: Optional[GetSessionConfig] = None,
   ) -> Optional[Session]:
+    _validate_session_id(session_id)
     reasoning_engine_id = self._get_reasoning_engine_id(app_name)
     session_resource_name = (
         f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
@@ -256,14 +271,30 @@ async def list_sessions(
   async def delete_session(
       self, *, app_name: str, user_id: str, session_id: str
   ) -> None:
+    _validate_session_id(session_id)
     reasoning_engine_id = self._get_reasoning_engine_id(app_name)
+    session_resource_name = (
+        f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
+    )
 
     async with self._get_api_client() as api_client:
+      # Enforce ownership: delete_session otherwise ignores user_id entirely.
+      try:
+        existing = await api_client.agent_engines.sessions.get(
+            name=session_resource_name
+        )
+      except ClientError as e:
+        if e.code == 404:
+          return
+        raise
+      if existing.user_id != user_id:
+        raise ValueError(
+            f'Session {session_id} does not belong to user {user_id}.'
+        )
+
       try:
         await api_client.agent_engines.sessions.delete(
-            name=(
-                f'reasoningEngines/{reasoning_engine_id}/sessions/{session_id}'
-            ),
+            name=session_resource_name,
         )
       except Exception as e:
         logger.error('Error deleting session %s: %s', session_id, e)
@@ -274,6 +305,7 @@ async def append_event(self, session: Session, event: Event) -> Event:
     # Update the in-memory session.
     await super().append_event(session=session, event=event)
 
+    _validate_session_id(session.id)
     reasoning_engine_id = self._get_reasoning_engine_id(session.app_name)
 
     # Build config (Monolithic approach)