Use versioned Redis keys for session identity

docs/guides/integrations/redis/redis_session_service/index.md

@@ -123,6 +123,11 @@ service = RedisSessionService(redis_client=shared_client)
 RedisSessionService requires the `redis` Python package to be installed. You
 can install it using `pip install google-adk[redis]`.
 
+Native Redis Cluster is not supported. When an event updates `app:` or `user:`
+state, the service watches the session and state keys in one transaction. These
+keys are not guaranteed to share a Cluster hash slot, so the transaction can
+fail with `CROSSSLOT`. Use a standalone Redis instance for this service.
+
 State keys prefixed with `temp:` are treated as transient. While they are
 available in the session object immediately after creation or an update, they
 are not persisted to Redis and will disappear when the session is reloaded

src/google/adk/integrations/redis/README.md

@@ -102,15 +102,22 @@ client = redis_asyncio.Redis.from_url(
 session_service = RedisSessionService(redis_client=client)
 ```
 
+Native Redis Cluster is not supported. When an event updates `app:` or `user:`
+state, the service watches the session and state keys in one transaction. These
+keys are not guaranteed to share a Cluster hash slot, so the transaction can
+fail with `CROSSSLOT`. Use a standalone Redis instance for this service.
+
 ## Key Schema & State Scoping
 
 `RedisSessionService` stores session data and state using the following Redis key structure:
 
 | Key Pattern | Data | Description |
 | :--- | :--- | :--- |
-| `{key_prefix}{app_name}:{user_id}:{session_id}` | JSON (`Session`) | Contains session ID, state, events list, and last update timestamp. Expired based on `ttl_seconds`. |
-| `{key_prefix}user_state:{app_name}:{user_id}` | JSON (`dict`) | Persisted user-scoped state (keys prefixed with `user:`). Shared across sessions for the same user. Expired based on `ttl_seconds`. |
-| `{key_prefix}app_state:{app_name}` | JSON (`dict`) | Persisted app-scoped state (keys prefixed with `app:`). Shared across all sessions and users for the app. Expired based on `ttl_seconds`. |
+| `{key_prefix}v2:s2_{b64url(app_name)}.{b64url(user_id)}.{b64url(session_id)}` | JSON (`Session`) | Contains session ID, state, events list, and last update timestamp. Expired based on `ttl_seconds`. |
+| `{key_prefix}v2:u2_{b64url(app_name)}.{b64url(user_id)}` | JSON (`dict`) | Persisted user-scoped state (keys prefixed with `user:`). Shared across sessions for the same user. Expired based on `ttl_seconds`. |
+| `{key_prefix}v2:a2_{b64url(app_name)}` | JSON (`dict`) | Persisted app-scoped state (keys prefixed with `app:`). Shared across all sessions and users for the app. Expired based on `ttl_seconds`. |
+
+`b64url(value)` is the unpadded URL-safe Base64 encoding of the UTF-8 value. With the default `key_prefix`, the new namespace is `adk:session:v2:`. These versioned keys are not compatible with keys written by earlier versions. This service does not read, migrate, or delete legacy entries; existing entries remain untouched and are not visible to this version. Plan a coordinated migration or cutover before upgrading. Legacy keys can expire under their configured TTL; if expiry is disabled, inventory and back up relevant entries before any manual cleanup, excluding the new `{key_prefix}v2:` keys. Do not flush a Redis instance shared with other applications.
 
 When events append state deltas:
 - `user:<key>` updates are synchronized to the user state key.

src/google/adk/integrations/redis/_redis_session_service.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+from base64 import urlsafe_b64encode
 import copy
 import json
 import logging
@@ -50,6 +51,11 @@
 logger = logging.getLogger("google_adk." + __name__)
 
 
+def _encode_key_component(value: str) -> str:
+  """Encodes an identity component as unpadded URL-safe Base64."""
+  return urlsafe_b64encode(value.encode("utf-8")).decode("ascii").rstrip("=")
+
+
 def _escape_glob(value: str) -> str:
   """Escapes Redis glob metacharacters so that `value` matches only itself."""
   return re.sub(r"([\\*?\[\]])", r"\\\1", value)
@@ -104,15 +110,32 @@ def _get_redis(self) -> Any:
 
   def _session_key(self, app_name: str, user_id: str, session_id: str) -> str:
     """Constructs the Redis key for a given session."""
-    return f"{self.config.key_prefix}{app_name}:{user_id}:{session_id}"
+    return (
+        f"{self.config.key_prefix}v2:s2_{_encode_key_component(app_name)}."
+        f"{_encode_key_component(user_id)}.{_encode_key_component(session_id)}"
+    )
 
   def _user_state_key(self, app_name: str, user_id: str) -> str:
     """Constructs the Redis key for user-scoped state."""
-    return f"{self.config.key_prefix}user_state:{app_name}:{user_id}"
+    return (
+        f"{self.config.key_prefix}v2:u2_{_encode_key_component(app_name)}."
+        f"{_encode_key_component(user_id)}"
+    )
 
   def _app_state_key(self, app_name: str) -> str:
     """Constructs the Redis key for app-scoped state."""
-    return f"{self.config.key_prefix}app_state:{app_name}"
+    return f"{self.config.key_prefix}v2:a2_{_encode_key_component(app_name)}"
+
+  @staticmethod
+  def _matches_session_identity(
+      session: Session, app_name: str, user_id: str, session_id: str
+  ) -> bool:
+    """Checks that a stored session belongs to the requested identity."""
+    return (
+        session.app_name == app_name
+        and session.user_id == user_id
+        and session.id == session_id
+    )
 
   @staticmethod
   def _merge_state(
@@ -271,6 +294,10 @@ async def get_session(
       return None
 
     session = Session.model_validate_json(raw)
+    if not self._matches_session_identity(
+        session, app_name, user_id, session_id
+    ):
+      return None
     app_state = json.loads(app_raw) if app_raw else {}
     user_state = json.loads(user_raw) if user_raw else {}
     session.state = self._merge_state(app_state, user_state, session.state)
@@ -293,12 +320,13 @@ async def list_sessions(
   ) -> ListSessionsResponse:
     """Lists sessions matching the given app_name and optional user_id."""
     client = self._get_redis()
-    escaped_app_name = _escape_glob(app_name)
     pattern = (
-        f"{self.config.key_prefix}{escaped_app_name}:{_escape_glob(user_id)}:*"
-        if user_id is not None
-        else f"{self.config.key_prefix}{escaped_app_name}:*"
+        f"{_escape_glob(self.config.key_prefix)}v2:"
+        f"s2_{_encode_key_component(app_name)}."
     )
+    if user_id is not None:
+      pattern += f"{_encode_key_component(user_id)}."
+    pattern += "*"
     app_raw = await client.get(self._app_state_key(app_name))
     app_state = json.loads(app_raw) if app_raw else {}
 
@@ -314,9 +342,14 @@ async def list_sessions(
       if raw:
         try:
           s = Session.model_validate_json(raw)
-          # Key segments are ':'-joined and the trailing wildcard spans ':', so
-          # a matched key can belong to another app or user whose id embeds a
-          # ':'. Confirm ownership from the stored session itself.
+          # A legacy key can match the v2 scan pattern; only list canonical keys.
+          expected_key = self._session_key(s.app_name, s.user_id, s.id)
+          if key != (
+              expected_key.encode("utf-8")
+              if isinstance(key, bytes)
+              else expected_key
+          ):
+            continue
           if s.app_name != app_name or (
               user_id is not None and s.user_id != user_id
           ):
@@ -359,11 +392,9 @@ async def append_event(self, session: Session, event: Event) -> Event:
     """Appends an event to the session and synchronizes state in Redis."""
     client = self._get_redis()
     key = self._session_key(session.app_name, session.user_id, session.id)
-    # Check before any write: the app/user state deltas below are stored under
-    # their own keys, and the session write would otherwise create a session
-    # that was never stored instead of failing like the other services.
     if not await client.exists(key):
       raise SessionNotFoundError(f"Session {session.id} not found.")
+
     event = await super().append_event(session, event)
     # Partial events are not persisted: they neither advance last_update_time
     # nor touch storage, matching every other session service backend.
@@ -376,8 +407,14 @@ async def append_event(self, session: Session, event: Event) -> Event:
     # last_update_time is the key list_sessions orders by.
     session.last_update_time = event.timestamp
 
-    # Sync app and user state deltas to their respective keys
+    # Collect state deltas before the transaction. State writes are queued in
+    # the same watched transaction as the session write so identity validation
+    # protects all persistent mutations.
+    app_delta: dict[str, Any] = {}
+    user_delta: dict[str, Any] = {}
     session_delta: dict[str, Any] = {}
+    app_key = None
+    user_key = None
     if event.actions and event.actions.state_delta:
       deltas = _session_util.extract_json_safe_state_delta(
           event.actions.state_delta
@@ -387,15 +424,10 @@ async def append_event(self, session: Session, event: Event) -> Event:
       session_delta = deltas.get("session", {})
 
       if app_delta:
-        await self._apply_scoped_state_delta(
-            client, self._app_state_key(session.app_name), app_delta
-        )
+        app_key = self._app_state_key(session.app_name)
+
       if user_delta:
-        await self._apply_scoped_state_delta(
-            client,
-            self._user_state_key(session.app_name, session.user_id),
-            user_delta,
-        )
+        user_key = self._user_state_key(session.app_name, session.user_id)
 
     # Prepare storage copy with ONLY session-scoped state (excluding app:, user:, temp:)
     session_only_state = _session_util.make_json_safe_state(
@@ -409,13 +441,22 @@ async def append_event(self, session: Session, event: Event) -> Event:
         events=session.events,
         last_update_time=session.last_update_time,
     )
+    watch_keys = [key]
+    if app_key:
+      watch_keys.append(app_key)
+    if user_key:
+      watch_keys.append(user_key)
 
     async def append_to_storage(pipe: Any) -> None:
       raw = await pipe.get(key)
       if raw:
         # The caller may hold a filtered view. Reload on every transaction
         # attempt so concurrent appends are preserved as well.
         stored_session = Session.model_validate_json(raw)
+        if not self._matches_session_identity(
+            stored_session, session.app_name, session.user_id, session.id
+        ):
+          raise SessionNotFoundError("Session not found.")
         storage_session.events = stored_session.events
         storage_session.state = stored_session.state
         storage_session.events.append(event)
@@ -424,12 +465,37 @@ async def append_to_storage(pipe: Any) -> None:
       else:
         storage_session.events = list(session.events)
         storage_session.state = dict(session_only_state)
+
+      app_state = None
+      if app_key:
+        app_raw = await pipe.get(app_key)
+        app_state = json.loads(app_raw) if app_raw else {}
+        app_state.update(app_delta)
+
+      user_state = None
+      if user_key:
+        user_raw = await pipe.get(user_key)
+        user_state = json.loads(user_raw) if user_raw else {}
+        user_state.update(user_delta)
+
       pipe.multi()
+      if app_key:
+        pipe.set(
+            app_key,
+            json.dumps(app_state),
+            ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None,
+        )
+      if user_key:
+        pipe.set(
+            user_key,
+            json.dumps(user_state),
+            ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None,
+        )
       pipe.set(
           key,
           storage_session.model_dump_json(),
           ex=self.config.ttl_seconds if self.config.ttl_seconds > 0 else None,
       )
 
-    await client.transaction(append_to_storage, key)
+    await client.transaction(append_to_storage, *watch_keys)
     return event

tests/unittests/integrations/redis/_fake_redis.py

@@ -85,21 +85,31 @@ async def delete(self, key: str) -> int:
     return 0
 
   async def transaction(
-      self, func: Callable[[Any], Awaitable[None]], key: str
-  ) -> list[bool | None]:
-    """Retries a queued write if the watched key changed or expired."""
+      self, func: Callable[[Any], Awaitable[None]], *keys: str
+  ) -> list[Any]:
+    """Retries queued writes if any watched key changed or expired."""
     while True:
-      self._is_expired(key)
-      version = self._versions.get(key, 0)
+      for key in keys:
+        self._is_expired(key)
+      versions = {key: self._versions.get(key, 0) for key in keys}
       pipe = Mock(get=self.get)
       await func(pipe)
-      self._is_expired(key)
-      if self._versions.get(key, 0) != version:
+      for key in keys:
+        self._is_expired(key)
+      if any(
+          self._versions.get(key, 0) != version
+          for key, version in versions.items()
+      ):
         continue
-      return [
+      results = [
+          await self.delete(*call.args, **call.kwargs)
+          for call in pipe.delete.call_args_list
+      ]
+      results.extend([
           await self.set(*call.args, **call.kwargs)
           for call in pipe.set.call_args_list
-      ]
+      ])
+      return results
 
   @staticmethod
   def _glob_match(pattern: str, key: str) -> bool:

tests/unittests/integrations/redis/test_redis_key_collision_probe.py

@@ -0,0 +1,91 @@
+# 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 collections.abc import AsyncGenerator
+
+from google.adk.agents import BaseAgent
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.events import Event
+from google.adk.integrations.redis import RedisSessionService
+from google.adk.runners import Runner
+from google.genai import types
+import pytest
+
+from ._fake_redis import FakeRedisAsync
+
+
+class _EchoAgent(BaseAgent):
+
+  async def _run_async_impl(
+      self, ctx: InvocationContext
+  ) -> AsyncGenerator[Event, None]:
+    yield Event(
+        invocation_id=ctx.invocation_id,
+        author=self.name,
+        content=types.Content(
+            role="model", parts=[types.Part(text="processed")]
+        ),
+    )
+
+
+@pytest.mark.asyncio
+async def test_runner_keeps_attacker_turn_out_of_victim_session():
+  service = RedisSessionService(redis_client=FakeRedisAsync())
+  await service.create_session(
+      app_name="billing",
+      user_id="tenant:victim",
+      session_id="fixed",
+      state={"private_record": "victim-only"},
+  )
+  await service.create_session(
+      app_name="billing",
+      user_id="tenant",
+      session_id="victim:fixed",
+      state={"attacker_record": "attacker-only"},
+  )
+  runner = Runner(
+      app_name="billing",
+      agent=_EchoAgent(name="echo"),
+      session_service=service,
+  )
+
+  _ = [
+      event
+      async for event in runner.run_async(
+          user_id="tenant",
+          session_id="victim:fixed",
+          new_message=types.Content(
+              role="user", parts=[types.Part(text="attacker turn")]
+          ),
+      )
+  ]
+
+  victim = await service.get_session(
+      app_name="billing", user_id="tenant:victim", session_id="fixed"
+  )
+  assert victim is not None
+  assert victim.state["private_record"] == "victim-only"
+  assert victim.events == []
+
+  attacker = await service.get_session(
+      app_name="billing", user_id="tenant", session_id="victim:fixed"
+  )
+  assert attacker is not None
+  assert attacker.state["attacker_record"] == "attacker-only"
+  assert [
+      part.text for event in attacker.events for part in event.content.parts
+  ] == [
+      "attacker turn",
+      "processed",
+  ]